diff --git a/crates/thetadatadx/src/config/fpss.rs b/crates/thetadatadx/src/config/fpss.rs index a19628c40..7dc8b9552 100644 --- a/crates/thetadatadx/src/config/fpss.rs +++ b/crates/thetadatadx/src/config/fpss.rs @@ -400,6 +400,18 @@ pub mod bounds { pub const PING_INTERVAL_MS: std::ops::RangeInclusive = 100..=300_000; /// Allowed range for [`super::StreamingConfig::io_read_slice_ms`], in milliseconds. pub const IO_READ_SLICE_MS: std::ops::RangeInclusive = 10..=500; + /// Allowed non-zero range for + /// [`super::StreamingConfig::data_watchdog_ms`], in milliseconds. + /// + /// `0` disables the watchdog and is handled separately; any enabled + /// value must fall inside this band. The floor matches the streaming + /// read-timeout floor ([`TIMEOUT_MS`]) because the watchdog is a + /// wall-clock backstop that sits *above* the read timeout — a value + /// below `timeout_ms` would fire before the read timeout and defeat + /// its purpose, so the `>= timeout_ms` invariant is also enforced in + /// `validate`. The ceiling bounds the longest a dead link can sit + /// undetected by the backstop at one hour. + pub const DATA_WATCHDOG_MS: std::ops::RangeInclusive = 100..=3_600_000; /// Allowed range for [`super::StreamingConfig::keepalive_idle_secs`], in seconds. pub const KEEPALIVE_IDLE_SECS: std::ops::RangeInclusive = 1..=7_200; /// Allowed range for [`super::StreamingConfig::keepalive_interval_secs`], in seconds. diff --git a/crates/thetadatadx/src/config/mod.rs b/crates/thetadatadx/src/config/mod.rs index 5b2d3cf9b..39eb88425 100644 --- a/crates/thetadatadx/src/config/mod.rs +++ b/crates/thetadatadx/src/config/mod.rs @@ -288,6 +288,29 @@ impl DirectConfig { to_i64(*streaming_bounds::IO_READ_SLICE_MS.end()), )); } + // `data_watchdog_ms` is a wall-clock backstop above the read + // timeout: `0` disables it, and any enabled value must sit inside + // its band and at or above `timeout_ms` so the backstop cannot + // fire before the read timeout it is meant to backstop. + if self.streaming.data_watchdog_ms != 0 { + if !streaming_bounds::DATA_WATCHDOG_MS.contains(&self.streaming.data_watchdog_ms) { + return Err(Error::config_out_of_range( + "streaming.data_watchdog_ms", + to_i64(self.streaming.data_watchdog_ms), + to_i64(*streaming_bounds::DATA_WATCHDOG_MS.start()), + to_i64(*streaming_bounds::DATA_WATCHDOG_MS.end()), + )); + } + if self.streaming.data_watchdog_ms < self.streaming.timeout_ms { + return Err(Error::config_invalid( + "streaming.data_watchdog_ms", + format!( + "data_watchdog_ms ({}) must be 0 (disabled) or >= timeout_ms ({})", + self.streaming.data_watchdog_ms, self.streaming.timeout_ms + ), + )); + } + } if !streaming_bounds::KEEPALIVE_IDLE_SECS.contains(&self.streaming.keepalive_idle_secs) { return Err(Error::config_out_of_range( "streaming.keepalive_idle_secs", @@ -767,6 +790,21 @@ mod config_file { concurrent_requests: usize, } + impl GrpcSection { + /// Upper ceiling for `[grpc] max_message_size_mb`, in megabytes. + /// + /// The inbound message size is a pre-allocated decode budget, so an + /// out-of-range value is a footgun in both directions: the + /// MB→byte conversion (`mb * 1024 * 1024`) overflows `usize` for + /// absurd inputs, and even a value that does not overflow commits + /// the channel to a buffer far beyond any legitimate response. The + /// production default is 4 MB; 64 MB leaves generous headroom for + /// the largest bulk historical chunk while keeping the budget + /// bounded. Values above this — or a `0` that would disable the + /// limit entirely — are rejected by name at load time. + const MAX_MESSAGE_SIZE_MB: usize = 64; + } + impl Default for GrpcSection { fn default() -> Self { let prod = DirectConfig::production(); @@ -903,7 +941,29 @@ mod config_file { // wins when present — including when set to the same number as // the default — and is inert when absent. let max_message_size = match cf.grpc.max_message_size_mb { - Some(mb) => mb * 1024 * 1024, + // The override is a pre-allocated decode budget. Reject a + // `0` (which would disable the limit) or an out-of-ceiling + // value up front, and compute the byte count with + // `checked_mul` so an absurd input is reported as a range + // error rather than wrapping `usize` into a tiny cap. + Some(mb) => { + if mb == 0 || mb > GrpcSection::MAX_MESSAGE_SIZE_MB { + return Err(Error::config_out_of_range( + "grpc.max_message_size_mb", + i64::try_from(mb).unwrap_or(i64::MAX), + 1, + i64::try_from(GrpcSection::MAX_MESSAGE_SIZE_MB).unwrap_or(i64::MAX), + )); + } + mb.checked_mul(1024 * 1024).ok_or_else(|| { + Error::config_out_of_range( + "grpc.max_message_size_mb", + i64::try_from(mb).unwrap_or(i64::MAX), + 1, + i64::try_from(GrpcSection::MAX_MESSAGE_SIZE_MB).unwrap_or(i64::MAX), + ) + })? + } None => cf.historical.max_message_size, }; @@ -1265,6 +1325,51 @@ mod tests { assert_eq!(config.historical.max_message_size, 4 * 1024 * 1024); } + #[test] + fn grpc_max_message_size_mb_at_ceiling_is_accepted() { + let toml = r#" + [grpc] + max_message_size_mb = 64 + "#; + let config = DirectConfig::from_toml_str(toml).unwrap(); + assert_eq!(config.historical.max_message_size, 64 * 1024 * 1024); + } + + #[test] + fn grpc_max_message_size_mb_above_ceiling_is_rejected() { + let toml = r#" + [grpc] + max_message_size_mb = 65 + "#; + let err = DirectConfig::from_toml_str(toml) + .expect_err("a value above the ceiling must be rejected"); + assert!(err.to_string().contains("max_message_size_mb"), "{err}"); + } + + #[test] + fn grpc_max_message_size_mb_zero_is_rejected() { + // `0` would disable the inbound-size limit entirely; it must be + // reported by name rather than silently uncapping the channel. + let toml = r#" + [grpc] + max_message_size_mb = 0 + "#; + let err = + DirectConfig::from_toml_str(toml).expect_err("a zero override must be rejected"); + assert!(err.to_string().contains("max_message_size_mb"), "{err}"); + } + + #[test] + fn grpc_max_message_size_mb_absurd_value_does_not_panic_or_wrap() { + // A value that would overflow the MB→byte conversion must + // surface as a range error, never a debug panic or a release + // wrap into a tiny garbage cap. + let toml = format!("[grpc]\nmax_message_size_mb = {}\n", usize::MAX); + let err = DirectConfig::from_toml_str(&toml) + .expect_err("an absurd value must be rejected, not wrapped"); + assert!(err.to_string().contains("max_message_size_mb"), "{err}"); + } + #[test] fn grpc_max_message_size_absent_keeps_historical_bytes() { // With no [grpc] override the canonical [historical] byte value @@ -1433,6 +1538,52 @@ mod tests { assert!(err.to_string().contains("io_read_slice_ms")); } + #[test] + fn validate_accepts_disabled_data_watchdog() { + let mut config = DirectConfig::production_defaults(); + config.streaming.data_watchdog_ms = 0; + let validated = config + .validate() + .expect("0 disables the watchdog and must validate"); + assert_eq!(validated.streaming.data_watchdog_ms, 0); + } + + #[test] + fn validate_rejects_data_watchdog_below_read_timeout() { + let mut config = DirectConfig::production_defaults(); + // Above the band floor but below timeout_ms — the backstop would + // fire before the read timeout it is meant to backstop. + config.streaming.timeout_ms = 5_000; + config.streaming.data_watchdog_ms = 1_000; + let err = config + .validate() + .expect_err("watchdog below timeout_ms must be rejected"); + let msg = err.to_string(); + assert!(msg.contains("data_watchdog_ms"), "{msg}"); + assert!(msg.contains("timeout_ms"), "{msg}"); + } + + #[test] + fn validate_rejects_data_watchdog_above_maximum() { + let mut config = DirectConfig::production_defaults(); + config.streaming.data_watchdog_ms = 7_200_000; + let err = config + .validate() + .expect_err("watchdog above the ceiling must be rejected"); + assert!(err.to_string().contains("data_watchdog_ms")); + } + + #[test] + fn validate_accepts_in_range_data_watchdog() { + let mut config = DirectConfig::production_defaults(); + config.streaming.timeout_ms = 3_000; + config.streaming.data_watchdog_ms = 60_000; + let validated = config + .validate() + .expect("an enabled watchdog at or above timeout_ms validates"); + assert_eq!(validated.streaming.data_watchdog_ms, 60_000); + } + #[test] fn validate_rejects_keepalive_out_of_range() { let mut config = DirectConfig::production_defaults(); diff --git a/crates/thetadatadx/src/util/ring.rs b/crates/thetadatadx/src/util/ring.rs index 4018d002d..e7e74b303 100644 --- a/crates/thetadatadx/src/util/ring.rs +++ b/crates/thetadatadx/src/util/ring.rs @@ -20,7 +20,22 @@ /// larger producers may trip the wrap fence on every fourth publish. pub const MIN_RING_SIZE: usize = 64; -/// Validate that `n` is a power of two no smaller than [`MIN_RING_SIZE`]. +/// Maximum ring buffer size accepted by [`check_ring_size`]. +/// +/// The ring is pre-allocated in full at construction, so the size is a +/// memory commitment, not a ceiling that is only reached under load. +/// `2^24` (16,777,216) slots bounds that commitment — at +/// `sizeof(Option)` per slot it is already a multi-hundred- +/// megabyte allocation, well above the shipped 131,072-slot default and +/// any realistic burst budget. Without an upper bound an absurd +/// power-of-two (`2^40`, say) passes the power-of-two and minimum checks +/// and the engine attempts to reserve terabytes up front, so the ceiling +/// is enforced and reported by name rather than left to a runtime +/// allocation failure. +pub const MAX_RING_SIZE: usize = 1 << 24; + +/// Validate that `n` is a power of two within +/// `[MIN_RING_SIZE, MAX_RING_SIZE]`. /// /// Returns `Ok(n)` on success; `Err(RingSizeError)` on failure. The /// error names the offending value and the nearest valid size so the @@ -29,7 +44,8 @@ pub const MIN_RING_SIZE: usize = 64; /// # Errors /// /// Returns [`RingSizeError::TooSmall`] when `n` is below -/// [`MIN_RING_SIZE`], or [`RingSizeError::NotPowerOfTwo`] when `n` is +/// [`MIN_RING_SIZE`], [`RingSizeError::TooLarge`] when `n` is above +/// [`MAX_RING_SIZE`], or [`RingSizeError::NotPowerOfTwo`] when `n` is /// not a power of two. pub fn check_ring_size(n: usize) -> Result { if n < MIN_RING_SIZE { @@ -38,6 +54,12 @@ pub fn check_ring_size(n: usize) -> Result { minimum: MIN_RING_SIZE, }); } + if n > MAX_RING_SIZE { + return Err(RingSizeError::TooLarge { + provided: n, + maximum: MAX_RING_SIZE, + }); + } if !n.is_power_of_two() { return Err(RingSizeError::NotPowerOfTwo { provided: n, @@ -57,6 +79,13 @@ pub enum RingSizeError { /// The minimum the validator accepts. minimum: usize, }, + /// `provided` is larger than [`MAX_RING_SIZE`]. + TooLarge { + /// The size the caller supplied. + provided: usize, + /// The maximum the validator accepts. + maximum: usize, + }, /// `provided` is not a power of two. `suggested` is the nearest /// valid power of two `>= provided` so the caller can pick the /// next viable budget without recomputing it. @@ -74,6 +103,9 @@ impl std::fmt::Display for RingSizeError { Self::TooSmall { provided, minimum } => { write!(f, "ring_size {provided} is below the minimum of {minimum}") } + Self::TooLarge { provided, maximum } => { + write!(f, "ring_size {provided} is above the maximum of {maximum}") + } Self::NotPowerOfTwo { provided, suggested, @@ -135,4 +167,39 @@ mod tests { assert!(msg.contains("16")); assert!(msg.contains("64")); } + + #[test] + fn accepts_maximum() { + assert_eq!(check_ring_size(MAX_RING_SIZE), Ok(MAX_RING_SIZE)); + } + + #[test] + fn shipped_default_is_under_maximum() { + // The default shipped in `config.default.toml`. + assert_eq!(check_ring_size(131_072), Ok(131_072)); + assert!(131_072 < MAX_RING_SIZE); + } + + #[test] + fn rejects_above_maximum() { + // A power of two one step above the ceiling must be rejected + // before the engine pre-allocates an absurd ring. + let oversized = MAX_RING_SIZE << 1; + let err = check_ring_size(oversized).unwrap_err(); + assert_eq!( + err, + RingSizeError::TooLarge { + provided: oversized, + maximum: MAX_RING_SIZE, + } + ); + } + + #[test] + fn error_message_names_maximum_on_too_large() { + let oversized = 1usize << 40; + let msg = check_ring_size(oversized).unwrap_err().to_string(); + assert!(msg.contains(&oversized.to_string())); + assert!(msg.contains(&MAX_RING_SIZE.to_string())); + } }