fix(flatfiles): emit strikes in dollars and bound the wire connection

crates/thetadatadx/src/config/flatfiles.rs

@@ -41,6 +41,27 @@ pub struct FlatFilesConfig {
     /// once, and an un-jittered ladder lands those retries in
     /// lockstep. Disable only for tests that assert exact timings.
     pub jitter: bool,
+
+    /// TCP + TLS connect timeout for one legacy-host attempt, in seconds.
+    ///
+    /// Bounds the time spent establishing the connection and running the
+    /// auth handshake against a single host before the attempt is
+    /// abandoned and the next host (or the retry ladder) takes over.
+    /// Without it a host that accepts the TCP connection but never
+    /// completes the TLS handshake would hang the request indefinitely.
+    /// Mirrors the historical (gRPC) channel's `connect_timeout_secs`.
+    pub connect_timeout_secs: u64,
+
+    /// Read timeout for a single FLAT_FILE response frame, in seconds.
+    ///
+    /// Bounds the wait for the next chunk once streaming has begun. A
+    /// server that stalls mid-stream — accepting the request but never
+    /// sending another frame or the FLAT_FILE_END terminator — would
+    /// otherwise block the download forever. On expiry the attempt fails
+    /// as a transient stall so the retry ladder reconnects on a fresh
+    /// session. Sized generously so a slow-but-progressing bulk transfer
+    /// is never cut off mid-chunk.
+    pub read_timeout_secs: u64,
 }
 
 impl FlatFilesConfig {
@@ -57,6 +78,15 @@ impl FlatFilesConfig {
             initial_backoff: Duration::from_secs(1),
             max_backoff: Duration::from_secs(30),
             jitter: true,
+            // Matches the historical (gRPC) channel connect bound; long
+            // enough for a TLS handshake behind NAT / VPN, short enough
+            // that a black-holed host fails over fast.
+            connect_timeout_secs: 10,
+            // A FLAT_FILE stream sends frequent chunks while it is making
+            // progress; 60 s between frames is far beyond any healthy
+            // inter-chunk gap, so a longer silence means the server has
+            // stalled and the attempt should fail over rather than hang.
+            read_timeout_secs: 60,
         }
     }
 

crates/thetadatadx/src/flatfiles/decoded_row.rs

@@ -87,9 +87,10 @@ impl FlatFileRow {
         Ok(Self {
             symbol: symbol.to_string(),
             expiration,
-            // Vendor fixed-point (1/1000 dollar) -> dollars at the
-            // typed boundary.
-            strike: strike.map(|s| f64::from(s) / 1000.0),
+            // Strikes are dollars on every client-facing surface; the
+            // shared conversion keeps CSV / JSONL / Arrow / typed-row in
+            // lockstep on the exact value.
+            strike: crate::flatfiles::index::strike_dollars(strike),
             right,
             fields,
         })

crates/thetadatadx/src/flatfiles/index.rs

@@ -104,6 +104,23 @@ pub(crate) fn parse_header(blob: &[u8]) -> Result<BlobHeader, Error> {
     })
 }
 
+/// Divisor that converts the vendor wire strike (tenths of a cent) into
+/// dollars. `210000` wire units ≡ `$210.00`. Invariant: strikes are
+/// always dollars on every client-facing surface (CSV, JSONL, Arrow, and
+/// the typed [`crate::flatfiles::decoded_row::FlatFileRow`]); the scaled
+/// integer wire form must never reach a caller. Centralised here so the
+/// scale lives in exactly one place across all output paths.
+pub(crate) const STRIKE_WIRE_PER_DOLLAR: f64 = 1000.0;
+
+/// Convert a wire strike (tenths of a cent) to dollars.
+///
+/// Returns `None` for stock entries, which carry no strike. Shared by
+/// every output path so CSV, JSONL, Arrow, and the typed row agree on the
+/// exact value for the same input.
+pub(crate) fn strike_dollars(strike: Option<i32>) -> Option<f64> {
+    strike.map(|s| f64::from(s) / STRIKE_WIRE_PER_DOLLAR)
+}
+
 /// One INDEX entry's contract key + DATA-section pointer.
 #[derive(Debug, Clone)]
 pub(crate) struct IndexEntry {
@@ -121,6 +138,14 @@ pub(crate) struct IndexEntry {
     pub(crate) block_end: u64,
 }
 
+impl IndexEntry {
+    /// This entry's strike in dollars, or `None` for stock entries. See
+    /// [`strike_dollars`] for the shared conversion every output path uses.
+    pub(crate) fn strike_dollars(&self) -> Option<f64> {
+        strike_dollars(self.strike)
+    }
+}
+
 /// Iterator over INDEX entries.
 ///
 /// Holds a borrowed slice of just the INDEX section bytes (i.e.

crates/thetadatadx/src/flatfiles/request.rs

@@ -193,7 +193,7 @@ pub async fn flatfile_request_raw_with_config(
     run_retry_loop(config, move |_attempt| {
         let creds = creds.clone();
         let path = output_for_attempt.clone();
-        async move { run_one_attempt(&creds, sec, req, date, &path).await }
+        async move { run_one_attempt(&creds, sec, req, date, &path, config).await }
     })
     .await
 }
@@ -257,6 +257,7 @@ async fn run_one_attempt(
     req: ReqType,
     date: &str,
     output_path: &Path,
+    config: &FlatFilesConfig,
 ) -> Result<PathBuf, Error> {
     // Build the host candidate list — try every (host, port) in priority
     // order, matching the vendor terminal's `MDDS_NJ_HOSTS` config.
@@ -268,7 +269,9 @@ async fn run_one_attempt(
         }
     }
 
-    let mut session = connect_and_login(&hosts, creds).await?;
+    let connect_timeout = std::time::Duration::from_secs(config.connect_timeout_secs.max(1));
+    let read_timeout = std::time::Duration::from_secs(config.read_timeout_secs.max(1));
+    let mut session = connect_and_login(&hosts, creds, connect_timeout).await?;
     tracing::debug!(target: "flatfiles", "authed against MDDS legacy: bundle={}", session.bundle);
 
     let request_id = next_id();
@@ -292,9 +295,22 @@ async fn run_one_attempt(
     // clean end-of-stream, no bookkeeping flag needed.
 
     loop {
-        let frame = match read_frame(&mut session.stream).await {
-            Ok(f) => f,
-            Err(e) => {
+        // Bound the wait for each frame. A server that stalls mid-stream
+        // — never sending the next chunk nor FLAT_FILE_END — would
+        // otherwise hang the download forever. On expiry, surface a
+        // transient stall so the retry ladder reconnects on a fresh
+        // session rather than blocking indefinitely.
+        let frame = match tokio::time::timeout(read_timeout, read_frame(&mut session.stream)).await
+        {
+            Err(_) => {
+                return Err(Error::FlatFilesUnavailable(
+                    FlatFilesUnavailableReason::StreamTruncated {
+                        bytes_received: total,
+                    },
+                ));
+            }
+            Ok(Ok(f)) => f,
+            Ok(Err(e)) => {
                 // Differentiate between EOF-after-some-data (truncation)
                 // and an outright protocol error.
                 if total > 0
@@ -442,6 +458,7 @@ mod tests {
             initial_backoff: std::time::Duration::from_millis(1),
             max_backoff: std::time::Duration::from_millis(4),
             jitter: false,
+            ..FlatFilesConfig::production_defaults()
         }
     }
 

crates/thetadatadx/src/flatfiles/session.rs

@@ -13,6 +13,7 @@
 //! from the METADATA payload.
 
 use std::sync::Arc;
+use std::time::Duration;
 
 use rustls::pki_types::ServerName;
 use rustls::ClientConfig;
@@ -172,24 +173,47 @@ pub(crate) async fn login(
 /// short-circuited: replaying it across every MDDS host is pointless,
 /// risks rate-limiting the account, and the original error already
 /// describes what the server objected to.
+///
+/// `connect_timeout` bounds the combined TCP + TLS handshake **and** auth
+/// exchange for a single host. A host that accepts the socket but never
+/// finishes the TLS handshake, or never returns the auth frames, would
+/// otherwise block the whole request forever; on expiry the attempt
+/// moves on to the next host (or surfaces a transient timeout the retry
+/// ladder reconnects on).
 pub(crate) async fn connect_and_login<'a>(
     hosts: &[MddsHost<'a>],
     creds: &Credentials,
+    connect_timeout: Duration,
 ) -> Result<AuthedSession, Error> {
     let mut last_err: Option<Error> = None;
     for host in hosts {
-        match connect_tls(MddsHost {
-            host: host.host,
-            port: host.port,
-        })
-        .await
-        {
-            Ok(mut stream) => match login(&mut stream, creds).await {
-                Ok(bundle) => return Ok(AuthedSession { stream, bundle }),
-                Err(e) if is_terminal_login_error(&e) => return Err(e),
-                Err(e) => last_err = Some(e),
-            },
-            Err(e) => last_err = Some(e),
+        let attempt = async {
+            let mut stream = connect_tls(MddsHost {
+                host: host.host,
+                port: host.port,
+            })
+            .await?;
+            let bundle = login(&mut stream, creds).await?;
+            Ok::<AuthedSession, Error>(AuthedSession { stream, bundle })
+        };
+        match tokio::time::timeout(connect_timeout, attempt).await {
+            Ok(Ok(session)) => return Ok(session),
+            Ok(Err(e)) if is_terminal_login_error(&e) => return Err(e),
+            Ok(Err(e)) => last_err = Some(e),
+            Err(_) => {
+                // Treated as a transient I/O timeout so the retry ladder
+                // reconnects; a stuck handshake on one host should not
+                // fail the whole request without a reconnect attempt.
+                last_err = Some(Error::Io(std::io::Error::new(
+                    std::io::ErrorKind::TimedOut,
+                    format!(
+                        "MDDS connect/login to {}:{} timed out after {}s",
+                        host.host,
+                        host.port,
+                        connect_timeout.as_secs()
+                    ),
+                )));
+            }
         }
     }
     Err(last_err.unwrap_or_else(|| Error::config_missing("mdds.hosts")))
@@ -224,4 +248,47 @@ mod tests {
         // first character of the JSON body must be '{'.
         assert_eq!(p[4], b'{');
     }
+
+    #[tokio::test]
+    async fn connect_and_login_times_out_on_unreachable_host() {
+        use crate::auth::Credentials;
+        use std::time::Instant;
+
+        // 192.0.2.1 is TEST-NET-1 (RFC 5737): guaranteed non-routable, so
+        // the TCP connect stalls until our timeout fires rather than
+        // failing fast. Without the connect bound this call would hang.
+        let hosts = [MddsHost {
+            host: "192.0.2.1",
+            port: 12_000,
+        }];
+        let creds = Credentials::new("user@example.com", "pw");
+        let budget = Duration::from_millis(150);
+        let started = Instant::now();
+        let result = connect_and_login(&hosts, &creds, budget).await;
+        let elapsed = started.elapsed();
+
+        // `AuthedSession` is not `Debug`; match the variants by hand so a
+        // success path fails the test without needing a `Debug` bound.
+        let err = match result {
+            Ok(_) => panic!("connect to a black-hole host must fail, not succeed"),
+            Err(e) => e,
+        };
+        // The bound must actually fire — a hung host cannot block forever.
+        assert!(
+            elapsed < Duration::from_secs(5),
+            "connect must abandon the host near its timeout, took {elapsed:?}"
+        );
+        // Classified transient so the retry ladder reconnects.
+        assert!(
+            error_is_transient_for_test(&err),
+            "connect timeout must be transient, got {err:?}"
+        );
+    }
+
+    /// Mirror of the request-layer transient classifier for the single
+    /// case this test exercises: a timed-out connect surfaces as
+    /// `Error::Io`, which the retry loop treats as transient.
+    fn error_is_transient_for_test(err: &Error) -> bool {
+        matches!(err, Error::Io(_))
+    }
 }