fix(config): bound message size, watchdog, and ring size at load time

crates/thetadatadx/src/config/fpss.rs

@@ -400,6 +400,18 @@ pub mod bounds {
     pub const PING_INTERVAL_MS: std::ops::RangeInclusive<u64> = 100..=300_000;
     /// Allowed range for [`super::StreamingConfig::io_read_slice_ms`], in milliseconds.
     pub const IO_READ_SLICE_MS: std::ops::RangeInclusive<u64> = 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<u64> = 100..=3_600_000;
     /// Allowed range for [`super::StreamingConfig::keepalive_idle_secs`], in seconds.
     pub const KEEPALIVE_IDLE_SECS: std::ops::RangeInclusive<u64> = 1..=7_200;
     /// Allowed range for [`super::StreamingConfig::keepalive_interval_secs`], in seconds.

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();

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<StreamEvent>)` 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<usize, RingSizeError> {
     if n < MIN_RING_SIZE {
@@ -38,6 +54,12 @@ pub fn check_ring_size(n: usize) -> Result<usize, RingSizeError> {
             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()));
+    }
 }