fix(fpss): bound per-session delta-decode state against non-conformant peers

crates/thetadatadx/src/flatfiles/index.rs

@@ -187,9 +187,20 @@ fn parse_one_entry(cur: &mut Cursor<&[u8]>, sec: SecType) -> Result<IndexEntry,
     let mut e = Cursor::new(&entry_buf[..]);
 
     let (root, exp, strike, right) = match sec {
-        SecType::Option | SecType::Index => {
-            // Index payload (when supported by the vendor) follows the
-            // option layout: root_len, root, exp, right, strike, date.
+        SecType::Index => {
+            // The flat-file service publishes no INDEX dataset, so the
+            // request layer ([`crate::flatfiles::types::flat_file_serves`])
+            // rejects every Index pair before any blob is fetched and this
+            // walker is never reached with an INDEX section. No
+            // vendor-confirmed INDEX entry layout exists, so rather than
+            // borrow the option layout and risk misparsing a future format,
+            // fail loudly with a typed unsupported error.
+            return Err(Error::decode_codec(
+                "flatfiles INDEX: index flat-file entries are not supported",
+            ));
+        }
+        SecType::Option => {
+            // Option layout: root_len, root, exp, right, strike, date.
             let root_len = read_u8(&mut e)? as usize;
             let mut root_bytes = vec![0u8; root_len];
             e.read_exact(&mut root_bytes)?;
@@ -418,6 +429,32 @@ mod tests {
         assert_eq!(entry.block_end, 100);
     }
 
+    #[test]
+    fn index_sec_type_is_rejected_as_unsupported() {
+        // The flat-file service publishes no INDEX dataset; the walker must
+        // reject an INDEX section with a typed error rather than borrow the
+        // option layout and risk misparsing a future vendor format.
+        let mut e = Vec::new();
+        e.push(3u8);
+        e.extend_from_slice(b"SPX");
+        e.extend_from_slice(&20_260_117i32.to_be_bytes());
+
+        let mut buf = Vec::new();
+        buf.extend_from_slice(&(e.len() as u16).to_be_bytes());
+        buf.extend_from_slice(&e);
+        buf.extend_from_slice(&0i32.to_be_bytes());
+        buf.extend_from_slice(&0i64.to_be_bytes());
+        buf.extend_from_slice(&100i64.to_be_bytes());
+
+        let mut iter = IndexIter::new(&buf, SecType::Index);
+        let err = iter.next().unwrap().unwrap_err();
+        let msg = err.to_string();
+        assert!(
+            msg.contains("index flat-file entries are not supported"),
+            "expected unsupported-index error, got: {msg}"
+        );
+    }
+
     #[test]
     fn over_framed_entry_size_is_rejected() {
         // entry_size claims more bytes than the contract block decodes;

crates/thetadatadx/src/fpss/decode.rs

@@ -1963,4 +1963,64 @@ mod tests {
             other => panic!("expected Data(Quote) for in-range price_type, got {other:?}"),
         }
     }
+
+    /// A non-conformant peer that streams ever-incrementing contract ids
+    /// within a single session (no START/STOP boundary) must not grow the
+    /// delta-decode maps without limit. Feeding well past the per-session
+    /// cap keeps every map bounded at or below the cap.
+    #[test]
+    fn delta_state_bounds_unbounded_distinct_contract_ids() {
+        let mut delta_state = DeltaState::new();
+        let mut out: TickFields = [0; crate::fpss::delta::MAX_DATA_FIELDS];
+
+        // Trade frame: 1 contract_id + 8 data fields (dev-server format).
+        let n = DeltaState::SESSION_CONTRACT_CAP + 5_000;
+        for id in 0..n as i32 {
+            let payload = encode_fit_row(&[id, 34_200_000, 0, 50, 6, 5_500_000, 57, 6, 20_250_428]);
+            let res = delta_state.decode_tick(
+                StreamMsgType::Trade as u8,
+                &payload,
+                TRADE_FIELDS,
+                &mut out,
+            );
+            assert!(res.is_some(), "every distinct-id absolute tick decodes");
+        }
+
+        let (prev, ohlcvc, field_counts) = delta_state.state_sizes();
+        let cap = DeltaState::SESSION_CONTRACT_CAP;
+        assert!(
+            prev <= cap,
+            "prev map must stay bounded by the session cap: {prev} > {cap}"
+        );
+        assert!(
+            field_counts <= cap,
+            "field_counts map must stay bounded by the session cap: {field_counts} > {cap}"
+        );
+        // `decode_tick` never inserts into `ohlcvc`, and the reset clears it
+        // alongside the other maps, so it stays empty here.
+        assert_eq!(ohlcvc, 0, "ohlcvc untouched by trade decode");
+    }
+
+    /// A normal session that stays under the cap retains every distinct
+    /// contract id: the bound never trips and no baseline is reset.
+    #[test]
+    fn delta_state_normal_session_retains_all_contracts() {
+        let mut delta_state = DeltaState::new();
+        let mut out: TickFields = [0; crate::fpss::delta::MAX_DATA_FIELDS];
+
+        let n = 500i32; // a generous live universe, well under the cap.
+        for id in 0..n {
+            let payload = encode_fit_row(&[id, 34_200_000, 0, 50, 6, 5_500_000, 57, 6, 20_250_428]);
+            delta_state
+                .decode_tick(StreamMsgType::Trade as u8, &payload, TRADE_FIELDS, &mut out)
+                .expect("absolute tick decodes");
+        }
+
+        let (prev, _ohlcvc, field_counts) = delta_state.state_sizes();
+        assert_eq!(prev, n as usize, "every distinct id retained in prev");
+        assert_eq!(
+            field_counts, n as usize,
+            "every distinct id retained in field_counts"
+        );
+    }
 }

crates/thetadatadx/src/fpss/delta.rs

@@ -54,6 +54,16 @@ pub struct DeltaState {
     /// Actual data field count from the first absolute tick for each
     /// `(msg_type, contract_id)`. The dev server sends 8-field trades (simple
     /// format) while production sends 16-field trades (extended format).
+    ///
+    /// The width is fixed at the first absolute row and not revised by later
+    /// rows. This is correct because the trade-format width is a server-wide
+    /// constant for the session, not a per-row property: every row a given
+    /// peer emits carries the same field count. Subsequent rows for the same
+    /// `(msg_type, contract_id)` are FIT delta rows whose changed-field count
+    /// is smaller than the full width, so they cannot be used to re-derive it.
+    /// The decode buffer is always the full [`MAX_DATA_FIELDS`] width with
+    /// unused slots zero-filled, so a stale width can only under-report which
+    /// trailing fields a caller reads, never read out of bounds.
     field_counts: HashMap<(u8, i32), usize>,
     /// Timestamp of last STOP (market close) signal. Used to suppress
     /// "unknown `contract_id`" warnings for 5 seconds after STOP;
@@ -65,6 +75,18 @@ pub struct DeltaState {
 /// are suppressed (stale ticks are expected during market-close teardown).
 const STOP_SUPPRESS_DURATION: Duration = Duration::from_secs(5);
 
+/// Upper bound on the number of distinct `(msg_type, contract_id)` rows the
+/// delta-decode maps retain within a single session (between START/STOP
+/// boundaries). A conformant session resets these maps at every session
+/// boundary, so they hold only the live universe — at most a few hundred
+/// `(msg_type, contract_id)` rows for the widest subscription. This cap sits
+/// well above any legitimate session yet keeps a non-conformant peer that
+/// streams ever-incrementing contract ids without a STOP from growing the
+/// maps without limit: on overflow the maps are cleared once and re-seeded
+/// from the current tick, trading a one-time delta-baseline reset (the next
+/// row for each contract decodes as absolute) for a hard memory ceiling.
+const MAX_SESSION_CONTRACT_ROWS: usize = 8192;
+
 impl DeltaState {
     #[doc(hidden)]
     pub fn new() -> Self {
@@ -181,7 +203,25 @@ impl DeltaState {
                 tick_n,
             );
         } else {
-            // First absolute tick: record the actual field count.
+            // First absolute tick for this `(msg_type, contract_id)`.
+            //
+            // Bound the per-session state: a conformant peer resets these
+            // maps at every START/STOP boundary, so an unbounded distinct-id
+            // count means the peer never signalled a session boundary. Clear
+            // once and re-seed from this tick so the maps cannot grow without
+            // limit. The current row decodes as absolute (no `prev`), which is
+            // exactly the post-reset baseline, so no value is corrupted.
+            if self.prev.len() >= MAX_SESSION_CONTRACT_ROWS {
+                tracing::warn!(
+                    rows = self.prev.len(),
+                    cap = MAX_SESSION_CONTRACT_ROWS,
+                    "delta-decode state exceeded per-session contract cap; resetting baselines (peer streamed an unbounded distinct-contract universe without a session boundary)"
+                );
+                self.prev.clear();
+                self.ohlcvc.clear();
+                self.field_counts.clear();
+            }
+            // Record the actual field count for the (possibly re-seeded) key.
             self.field_counts.insert(key, tick_n);
         }
 
@@ -194,4 +234,15 @@ impl DeltaState {
         let data_fields = *self.field_counts.get(&key).unwrap_or(&expected_fields);
         Some((contract_id, data_fields))
     }
+
+    /// Per-session distinct-contract cap, exposed for boundedness tests.
+    #[cfg(test)]
+    pub(super) const SESSION_CONTRACT_CAP: usize = MAX_SESSION_CONTRACT_ROWS;
+
+    /// Distinct-row counts of the three per-session maps, exposed for
+    /// boundedness tests.
+    #[cfg(test)]
+    pub(super) fn state_sizes(&self) -> (usize, usize, usize) {
+        (self.prev.len(), self.ohlcvc.len(), self.field_counts.len())
+    }
 }