diff --git a/crates/thetadatadx/src/config/mdds.rs b/crates/thetadatadx/src/config/mdds.rs index 897ccb8e9..ed990d5ef 100644 --- a/crates/thetadatadx/src/config/mdds.rs +++ b/crates/thetadatadx/src/config/mdds.rs @@ -79,6 +79,25 @@ pub struct HistoricalConfig { /// cadence. pub connect_timeout_secs: u64, + /// Default per-request deadline for historical (gRPC) queries, in + /// seconds. `0` disables the default (no deadline unless the caller + /// sets one). + /// + /// A server that holds the HTTP/2 stream open while sending no + /// chunks would otherwise hang `collect_stream` / `stream(...)` + /// indefinitely: the gRPC keepalive PING only detects a fully dead + /// peer, not a live-but-silent one. This default bounds every + /// request that did not call `with_deadline(...)`, so a stalled + /// stream resolves to `Error::Timeout` instead of blocking forever. + /// + /// Per-call control overrides this: `with_deadline(Duration)` sets a + /// shorter or longer bound, and `with_deadline(Duration::ZERO)` + /// opts a single request out of any deadline. + /// + /// Default `300s` (5 min) — comfortably above the slowest realistic + /// multi-million-row bulk pull while still bounding a wedged stream. + pub request_timeout_secs: u64, + /// Estimated-bytes threshold above which the buffered `.await` /// path on a `parsed_endpoint!` builder emits a single /// `tracing::warn!` event suggesting `.stream(handler)` for the @@ -138,6 +157,11 @@ impl HistoricalConfig { window_size_kb: 64, connection_window_size_kb: 64, connect_timeout_secs: 10, + // 5 min — bounds a server that holds the stream open while + // sending no chunks (h2 keepalive only catches a fully dead + // peer). Sits above the slowest realistic bulk pull; + // `with_deadline(Duration::ZERO)` opts a single request out. + request_timeout_secs: 300, // 100 MiB — empirically catches bulk pulls (multi-million // row option-chain or multi-day backfill responses) while // staying silent on ad-hoc single-day quote / OHLC pulls diff --git a/crates/thetadatadx/src/fpss/connection.rs b/crates/thetadatadx/src/fpss/connection.rs index 940c076b4..28d6d5c87 100644 --- a/crates/thetadatadx/src/fpss/connection.rs +++ b/crates/thetadatadx/src/fpss/connection.rs @@ -223,7 +223,8 @@ pub(crate) fn order_hosts( /// 2. `TCP_NODELAY = true` /// 3. `SO_KEEPALIVE` armed per `keepalive` /// 4. Read timeout set to `read_timeout` -/// 5. TLS handshake pinned to the FPSS SPKI +/// 5. Write timeout set to `write_timeout` +/// 6. TLS handshake pinned to the FPSS SPKI /// /// # Errors /// @@ -232,6 +233,7 @@ pub(crate) fn connect_to_servers( servers: &[(&str, u16)], connect_timeout: Duration, read_timeout: Duration, + write_timeout: Duration, keepalive: TcpKeepaliveSpec, ) -> Result<(FpssStream, String), crate::error::Error> { ensure_rustls_crypto_provider(); @@ -241,7 +243,14 @@ pub(crate) fn connect_to_servers( let addr = format!("{host}:{port}"); tracing::debug!(server = %addr, "attempting FPSS connection"); - match try_connect(host, port, connect_timeout, read_timeout, keepalive) { + match try_connect( + host, + port, + connect_timeout, + read_timeout, + write_timeout, + keepalive, + ) { Ok(stream) => { tracing::info!(server = %addr, "FPSS connected"); return Ok((stream, addr)); @@ -318,12 +327,14 @@ fn arm_keepalive(tcp: &TcpStream, spec: TcpKeepaliveSpec) { /// 2. `set_nodelay(true)` /// 3. `SO_KEEPALIVE` per the configured schedule /// 4. `set_read_timeout` -/// 5. Blocking TLS handshake via rustls `StreamOwned` +/// 5. `set_write_timeout` +/// 6. Blocking TLS handshake via rustls `StreamOwned` fn try_connect( host: &str, port: u16, connect_timeout: Duration, read_timeout: Duration, + write_timeout: Duration, keepalive: TcpKeepaliveSpec, ) -> Result { let addr = format!("{host}:{port}"); @@ -358,6 +369,17 @@ fn try_connect( // Read timeout. tcp.set_read_timeout(Some(read_timeout))?; + // Write timeout. The first write (CREDENTIALS) drives the lazy TLS + // handshake, and steady-state ping/subscribe writes can otherwise + // block indefinitely against a peer whose receive window has stalled + // (alive enough to ACK at the kernel but not draining the socket). + // `connect_timeout` only bounds the SYN/ACK, so an unbounded write + // would wedge the I/O thread past that budget. The bound persists for + // the life of the socket via `SO_SNDTIMEO`, so a write `TimedOut` + // surfaces as a fatal I/O error and the caller reconnects, mirroring + // the read-timeout liveness contract. + tcp.set_write_timeout(Some(write_timeout))?; + // TLS handshake (blocking) using rustls with webpki root certificates. let server_name = ServerName::try_from(host.to_owned()).map_err(|e| crate::error::Error::Fpss { @@ -447,7 +469,14 @@ mod tests { let connect_timeout = Duration::from_millis(150); let read_timeout = Duration::from_millis(10_000); let start = std::time::Instant::now(); - let res = connect_to_servers(&servers, connect_timeout, read_timeout, test_keepalive()); + let write_timeout = Duration::from_millis(10_000); + let res = connect_to_servers( + &servers, + connect_timeout, + read_timeout, + write_timeout, + test_keepalive(), + ); let elapsed = start.elapsed(); assert!(res.is_err(), "unroutable host must fail to connect"); assert!( @@ -472,6 +501,39 @@ mod tests { ); } + /// Both the read and write timeouts round-trip onto a real socket. + /// + /// `try_connect` sets `SO_RCVTIMEO` and `SO_SNDTIMEO` before the TLS + /// handshake. The handshake itself needs a live FPSS peer, so this + /// asserts the load-bearing socket-option contract directly on a + /// loopback socket: an unbounded write would let a stalled-receiver + /// peer wedge the I/O thread past `connect_timeout`, so the write + /// timeout must actually land on the kernel socket. + #[test] + fn read_and_write_timeouts_arm_on_loopback_socket() { + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind loopback"); + let addr = listener.local_addr().expect("local addr"); + let tcp = TcpStream::connect(addr).expect("connect loopback"); + + let read_timeout = Duration::from_millis(7_000); + let write_timeout = Duration::from_millis(3_000); + tcp.set_read_timeout(Some(read_timeout)) + .expect("set read timeout"); + tcp.set_write_timeout(Some(write_timeout)) + .expect("set write timeout"); + + assert_eq!( + tcp.read_timeout().expect("read SO_RCVTIMEO"), + Some(read_timeout), + "read timeout must round-trip onto the socket" + ); + assert_eq!( + tcp.write_timeout().expect("read SO_SNDTIMEO"), + Some(write_timeout), + "write timeout must round-trip onto the socket" + ); + } + fn production_hosts() -> Vec<(String, u16)> { vec![ ("nj-a.thetadata.us".to_string(), 20000), diff --git a/crates/thetadatadx/src/fpss/io_loop/mod.rs b/crates/thetadatadx/src/fpss/io_loop/mod.rs index e716567da..e9e92044a 100644 --- a/crates/thetadatadx/src/fpss/io_loop/mod.rs +++ b/crates/thetadatadx/src/fpss/io_loop/mod.rs @@ -791,7 +791,17 @@ where .iter() .map(|(host, port)| (host.as_str(), *port)) .collect(); - connection::connect_to_servers(&ordered, connect_timeout, read_timeout, keepalive) + // The write timeout shares the read timeout's budget: both + // bound a single unacknowledged transport operation during the + // reconnect window, so the re-auth write cannot wedge the I/O + // thread against a peer that has stopped draining its socket. + connection::connect_to_servers( + &ordered, + connect_timeout, + read_timeout, + read_timeout, + keepalive, + ) }; let (mut new_stream, new_addr) = match new_stream { diff --git a/crates/thetadatadx/src/fpss/io_loop/ping.rs b/crates/thetadatadx/src/fpss/io_loop/ping.rs index 81ee7d35c..95a6c4ff5 100644 --- a/crates/thetadatadx/src/fpss/io_loop/ping.rs +++ b/crates/thetadatadx/src/fpss/io_loop/ping.rs @@ -55,12 +55,27 @@ pub(in crate::fpss) fn ping_loop( code: StreamMsgType::Ping, payload: ping_payload.clone(), }; - // Blocking send on the bounded channel: the heartbeat takes natural - // backpressure if the I/O thread is momentarily behind, rather than - // dropping a ping. `send` only errors once the receiver hangs up. - if cmd_tx.send(cmd).is_err() { - // I/O thread has exited - break; + // Non-blocking send on the bounded channel, matching the rest of + // the control plane (subscribe / unsubscribe / shutdown all use + // `try_send`). The heartbeat is an idempotent liveness signal, + // not a state-carrying frame: a momentarily full channel means + // the I/O thread is already draining a large backlog of outbound + // writes, so the connection is demonstrably alive and the next + // ping fires one interval later — well inside any server-side + // ping deadline. Skipping a ping under that backpressure is + // therefore safe, and it keeps a blocking heartbeat from pinning + // the consumer/Drop path for up to one interval while the channel + // drains. A hung-up receiver (I/O thread exited) still ends the + // loop. + match cmd_tx.try_send(cmd) { + Ok(()) => {} + Err(std_mpsc::TrySendError::Full(_)) => { + // I/O thread is behind; drop this beat and try the next. + } + Err(std_mpsc::TrySendError::Disconnected(_)) => { + // I/O thread has exited. + break; + } } thread::sleep(interval); diff --git a/crates/thetadatadx/src/fpss/mod.rs b/crates/thetadatadx/src/fpss/mod.rs index 6a2070e57..17fe601ef 100644 --- a/crates/thetadatadx/src/fpss/mod.rs +++ b/crates/thetadatadx/src/fpss/mod.rs @@ -1138,8 +1138,18 @@ impl StreamingClient { .collect(); let connect_timeout = Duration::from_millis(connect_timeout_ms); let read_timeout = Duration::from_millis(read_timeout_ms); - let (stream, server_addr) = - connection::connect_to_servers(&borrowed, connect_timeout, read_timeout, keepalive)?; + // The write deadline bounds the credentials write that drives the + // lazy TLS handshake and every steady-state ping/subscribe write. + // It shares the read timeout's budget: both bound a single + // unacknowledged transport operation during the connect window. + let write_timeout = read_timeout; + let (stream, server_addr) = connection::connect_to_servers( + &borrowed, + connect_timeout, + read_timeout, + write_timeout, + keepalive, + )?; Self::connect_with_stream(connection::ConnectWithStreamArgs { creds, stream, @@ -1692,8 +1702,19 @@ impl StreamingClient { /// call this at entry; the one-shot /// `AtomicBool` keeps the affinity syscall off the per-event path /// when `next_event` is driven one event at a time. A `None` - /// `consumer_cpu` short-circuits to nothing. + /// `consumer_cpu` short-circuits the affinity pin. + /// + /// Recording the consumer thread id is the single choke point that + /// arms the `Drop` self-join guard in production: every drain path + /// routes through here, so the thread that drives the ring is + /// captured exactly once. Without it `Drop` could join the I/O + /// handle inline on the dispatcher thread a user callback called + /// back into — that thread is the one cleanup must complete on, so + /// the inline join would self-deadlock. The capture happens before + /// the `consumer_cpu` short-circuit so the guard arms regardless of + /// whether CPU pinning is configured. fn pin_consumer_once(&self) { + self.record_consumer_thread(); if self.consumer_cpu.is_none() { return; } @@ -1703,6 +1724,17 @@ impl StreamingClient { affinity::pin_consumer_thread(self.consumer_cpu); } + /// Capture the draining thread's id exactly once so the `Drop` + /// self-join guard can detect a `Drop` running on the consumer + /// thread. Idempotent: the first drain wins, later calls are a cheap + /// `OnceLock` read. Separated from the CPU pin so the non-blocking + /// [`Self::poll_batch`] primitive arms the guard without inheriting + /// the affinity pin the blocking drains apply. + fn record_consumer_thread(&self) { + self.consumer_thread_id + .get_or_init(|| thread::current().id()); + } + /// Block the calling thread until the next event is available or /// the ring shuts down. /// @@ -1817,6 +1849,11 @@ impl StreamingClient { /// `&StreamEvent` handed to `on_event` is a zero-copy borrow into the /// ring slot, valid only for that call. pub fn poll_batch(&self, mut on_event: impl FnMut(&StreamEvent)) -> PollOutcome { + // Arm the self-join guard from this drain path too: a caller + // driving `poll_batch` in its own loop is the consumer thread, + // and a `Drop` reached from inside `on_event` must detach its + // join rather than block this thread on its own termination. + self.record_consumer_thread(); let Ok(mut guard) = self.poller_state.lock() else { return PollOutcome::Shutdown; }; @@ -1958,7 +1995,7 @@ impl StreamingClient { } /// Drain events through `on_event`, applying a caller-supplied - /// [`disruptor::wait_strategies::WaitStrategy`] on each momentarily + /// [`crate::streaming::wait::WaitStrategy`] on each momentarily /// empty ring instead of the configured /// [`crate::StreamingWaitStrategy`] preset. /// @@ -1967,7 +2004,10 @@ impl StreamingClient { /// `wait_park_us` knobs cover the common latency-vs-CPU points across /// every binding, but a Rust caller with an exotic backoff (e.g. an /// adaptive PID-controlled park, or a strategy that coordinates with - /// another subsystem) can supply any `W: WaitStrategy` here. + /// another subsystem) can supply any `W: WaitStrategy` here. Use a + /// preset from [`crate::streaming::wait`] (e.g. + /// [`crate::streaming::wait::BusySpin`]) or implement the trait on + /// your own type. /// /// `W` is monomorphised into the loop, so the per-poll cost is the /// caller's `wait_for` body with no indirection — identical @@ -1986,7 +2026,7 @@ impl StreamingClient { /// numeric spin / yield / park tuning, which every binding exposes. pub fn for_each_with_wait_strategy(&self, mut on_event: F, strategy: W) where - W: disruptor::wait_strategies::WaitStrategy, + W: crate::streaming::wait::WaitStrategy, F: FnMut(&StreamEvent), { self.pin_consumer_once(); @@ -2645,6 +2685,16 @@ impl StreamingClient { (client, producer) } + /// The consumer thread id recorded by the drain paths, if any drain + /// has run. `None` before the first `next_event` / `for_each*` / + /// `poll_batch` call. Exists so a test can assert the `Drop` + /// self-join guard is armed by a real drain path rather than only by + /// the test-only `for_self_join_test` setter. + #[cfg(test)] + pub(in crate::fpss) fn recorded_consumer_thread_id(&self) -> Option { + self.consumer_thread_id.get().copied() + } + /// Send a command to the I/O thread over the bounded control channel. /// /// Uses a non-blocking `try_send` so a public `&self` caller is never @@ -3484,4 +3534,93 @@ mod ring_occupancy_tests { client.shutdown(); } + + /// A real drain path arms the `Drop` self-join guard. + /// + /// In production the only thing that records the consumer thread id + /// is the drain entry point (`pin_consumer_once` for the blocking + /// drains, `record_consumer_thread` for `poll_batch`). If that + /// recording regresses, `Drop` reads a `None` thread id, never + /// detects a `Drop` running on the consumer thread, and joins the + /// I/O handle inline — a self-join deadlock when a user callback + /// drops the last client handle. This pins the recording to a real + /// drain call rather than the test-only setter: before any drain the + /// id is unset; after `poll_batch` on this thread it equals this + /// thread, so the guard's `consumer_id == Some(cur)` branch fires. + #[test] + fn drain_path_records_consumer_thread_id_for_self_join_guard() { + let (client, mut producer) = StreamingClient::for_ring_occupancy_test(64); + + assert_eq!( + client.recorded_consumer_thread_id(), + None, + "no drain has run yet; the guard must be unarmed" + ); + + assert!( + publish_one(&mut producer), + "fresh ring must accept a publish" + ); + let mut delivered = 0usize; + let _ = client.poll_batch(|_event| delivered += 1); + + assert_eq!( + client.recorded_consumer_thread_id(), + Some(std::thread::current().id()), + "poll_batch must record the draining thread so Drop detects a self-join" + ); + } + + /// The bring-your-own-strategy drain is reachable through the + /// crate-owned wait-strategy re-export, so a Rust caller never names + /// the underlying ring crate. Dropping the producer terminates the + /// drain; `BusySpin`'s `wait_for` is a no-op, so the loop spins only + /// until the ring reports shutdown. + #[test] + fn for_each_with_wait_strategy_accepts_crate_owned_preset() { + use crate::streaming::wait::BusySpin; + + let (client, mut producer) = StreamingClient::for_ring_occupancy_test(64); + for _ in 0..3 { + assert!( + publish_one(&mut producer), + "fresh ring must accept a publish" + ); + } + // Drop the producer so the poller observes terminal shutdown once + // the three published events drain, and the loop returns. + drop(producer); + + let mut delivered = 0usize; + client.for_each_with_wait_strategy(|_event| delivered += 1, BusySpin); + + assert_eq!( + delivered, 3, + "every published event must be delivered before shutdown returns" + ); + } + + /// The blocking `next_event` drain also arms the guard, covering the + /// `pin_consumer_once` entry point shared by every blocking drain. + #[test] + fn next_event_records_consumer_thread_id_for_self_join_guard() { + let (client, mut producer) = StreamingClient::for_ring_occupancy_test(64); + + assert_eq!(client.recorded_consumer_thread_id(), None); + + assert!( + publish_one(&mut producer), + "fresh ring must accept a publish" + ); + let event = client + .next_event() + .expect("staging mutex must not be poisoned"); + assert!(event.is_some(), "one event must be delivered"); + + assert_eq!( + client.recorded_consumer_thread_id(), + Some(std::thread::current().id()), + "next_event must record the draining thread so Drop detects a self-join" + ); + } } diff --git a/crates/thetadatadx/src/lib.rs b/crates/thetadatadx/src/lib.rs index 5a819f6f6..71f5405cd 100644 --- a/crates/thetadatadx/src/lib.rs +++ b/crates/thetadatadx/src/lib.rs @@ -268,6 +268,46 @@ pub use error::{ /// the crate root for callers that drive the batch loop directly. pub use fpss::PollOutcome; +/// Real-time streaming consumer surface. +pub mod streaming { + /// Consumer wait strategies for the streaming ring. + /// + /// When the consumer drains the ring faster than events arrive, it + /// must decide how to wait on a momentarily empty ring. For most + /// callers the [`crate::StreamingWaitStrategy`] preset enum plus the + /// `wait_spin_iters` / `wait_yield_iters` / `wait_park_us` numeric + /// knobs cover the full latency-versus-CPU spectrum, and that path is + /// the one every language binding exposes. + /// + /// A Rust caller that needs an exotic backoff the presets do not + /// model can instead supply any type implementing `WaitStrategy` to + /// [`crate::fpss::StreamingClient::for_each_with_wait_strategy`]. The + /// strategy is monomorphised into the drain loop, so the per-poll + /// cost is the caller's `wait_for` body with no indirection. + /// + /// `BusySpin` is the lowest-latency preset (a true busy spin); + /// `BusySpinWithSpinLoopHint` adds a `spin_loop` hint so the core + /// can save power or switch hyper-threads; `Sleep` parks the thread + /// for a fixed duration between polls. + /// + /// ```rust,ignore + /// use thetadatadx::streaming::wait::BusySpin; + /// + /// client.for_each_with_wait_strategy( + /// |event| { /* handle event */ }, + /// BusySpin, + /// ); + /// ``` + pub mod wait { + // VOCAB-OK: re-exporting the ring's wait-strategy surface under a + // crate-owned path so callers never name the underlying ring + // crate in their own `use` statements or trait bounds. + pub use disruptor::wait_strategies::{ + BusySpin, BusySpinWithSpinLoopHint, Sleep, WaitStrategy, + }; + } +} + // ─── Flat-file bulk pulls ───────────────────────────────────────────────────── /// Bulk flat-file downloads from ThetaData's flat-file distribution. diff --git a/crates/thetadatadx/src/mdds/macros.rs b/crates/thetadatadx/src/mdds/macros.rs index 61df5ea0b..cb6b802c0 100644 --- a/crates/thetadatadx/src/mdds/macros.rs +++ b/crates/thetadatadx/src/mdds/macros.rs @@ -31,6 +31,23 @@ /// /// Returns `Error::Timeout` when the deadline elapses; otherwise propagates /// whatever error the wrapped future resolves to. +/// Resolve the effective per-request deadline. +/// +/// An explicit `with_deadline(...)` always wins, including +/// `with_deadline(Duration::ZERO)` which normalizes to `None` at the +/// builder and opts the request out of any deadline. When the caller set +/// nothing (`explicit == None`), fall back to the configured +/// [`crate::config::HistoricalConfig::request_timeout_secs`] default so a +/// server holding the stream open without sending chunks cannot hang the +/// request indefinitely. A configured default of `0` disables the +/// fallback. +pub(crate) fn effective_deadline( + explicit: Option, + default_secs: u64, +) -> Option { + explicit.or_else(|| (default_secs > 0).then(|| std::time::Duration::from_secs(default_secs))) +} + pub(crate) async fn run_with_optional_deadline( deadline: Option, fut: F, @@ -689,6 +706,10 @@ macro_rules! parsed_endpoint { } = self; let _ = &client; $($($crate::mdds::validate::validate_date_required(&$date_arg)?;)+)? + let deadline = $crate::mdds::macros::effective_deadline( + deadline, + client.config().historical.request_timeout_secs, + ); $crate::mdds::macros::run_with_optional_deadline(deadline, async move { tracing::debug!(endpoint = stringify!($name), "gRPC streaming request"); metrics::counter!("thetadatadx.grpc.requests", "endpoint" => stringify!($name)).increment(1); @@ -863,6 +884,10 @@ macro_rules! parsed_endpoint { ); Ok(parsed) }; + let deadline = $crate::mdds::macros::effective_deadline( + deadline, + client.config().historical.request_timeout_secs, + ); $crate::mdds::macros::run_with_optional_deadline(deadline, inner).await }) } @@ -1074,6 +1099,54 @@ mod classify_error_tests { } } +#[cfg(test)] +mod effective_deadline_tests { + use super::effective_deadline; + use std::time::Duration; + + /// With no explicit `with_deadline(...)`, the configured default is + /// applied. This is the load-bearing assertion that a request which + /// set no deadline still gets one, so a server holding the stream + /// open without sending chunks cannot hang the call forever. + #[test] + fn applies_configured_default_when_caller_set_none() { + assert_eq!( + effective_deadline(None, 300), + Some(Duration::from_secs(300)) + ); + } + + /// An explicit deadline always wins over the configured default. + #[test] + fn explicit_deadline_overrides_default() { + assert_eq!( + effective_deadline(Some(Duration::from_secs(5)), 300), + Some(Duration::from_secs(5)) + ); + } + + /// `with_deadline(Duration::ZERO)` normalizes to `None` at the + /// builder, opting the request out of any deadline. A configured + /// default of `0` must NOT silently re-impose one on that request, + /// so a `0` default leaves the explicit opt-out intact. + #[test] + fn zero_default_disables_the_fallback() { + assert_eq!(effective_deadline(None, 0), None); + } + + /// The production default seeds a positive per-request deadline, so + /// the historical request path is bounded out of the box. + #[test] + fn production_default_is_positive() { + let cfg = crate::config::HistoricalConfig::production_defaults(); + assert!(cfg.request_timeout_secs > 0); + assert_eq!( + effective_deadline(None, cfg.request_timeout_secs), + Some(Duration::from_secs(cfg.request_timeout_secs)) + ); + } +} + #[cfg(test)] mod streaming_attempt_tests { //! Outcome routing for the streaming retry / refresh shell driven by diff --git a/docs-site/docs/migration/v12-to-v13.md b/docs-site/docs/migration/v12-to-v13.md index 1f749f8f3..c0b22164d 100644 --- a/docs-site/docs/migration/v12-to-v13.md +++ b/docs-site/docs/migration/v12-to-v13.md @@ -689,7 +689,7 @@ thetadatadx_config_set_consumer_cpu(cfg, 3); /* negative = unpinned */ Rust callers who need an exotic backoff can bring their own strategy through a zero-cost generic drain, which has no FFI / interpreter-lock cost because the per-poll wait is monomorphised into the loop: ```rust -use disruptor::wait_strategies::BusySpin; +use thetadatadx::streaming::wait::BusySpin; client.for_each_with_wait_strategy(|event| handle(event), BusySpin); ``` diff --git a/ffi/src/auth.rs b/ffi/src/auth.rs index 6ab4a8c72..2846d1241 100644 --- a/ffi/src/auth.rs +++ b/ffi/src/auth.rs @@ -2625,6 +2625,50 @@ pub unsafe extern "C" fn thetadatadx_config_get_warn_on_buffered_threshold_bytes }) } +/// Set the default per-request deadline (seconds) for historical queries +/// on a config handle. +/// +/// Bounds every request that did not call `with_deadline(...)`, so a +/// live-but-silent stream resolves to a timeout instead of blocking +/// forever (the gRPC keepalive PING only detects a fully dead peer). +/// `secs = 0` disables the default (no deadline unless the caller sets +/// one). Default `300`. +#[no_mangle] +pub unsafe extern "C" fn thetadatadx_config_set_request_timeout_secs( + config: *mut ThetaDataDxConfig, + secs: u64, +) { + ffi_boundary!((), { + let config = require_config_mut!(config); + config.inner.historical.request_timeout_secs = secs; + }) +} + +/// Read the current historical `request_timeout_secs` setting (default +/// `300`; `0` = no default deadline). +/// +/// Writes the configured value into `*out`. Returns `0` on success, +/// `-1` if either pointer is null. +#[no_mangle] +pub unsafe extern "C" fn thetadatadx_config_get_request_timeout_secs( + config: *const ThetaDataDxConfig, + out: *mut u64, +) -> i32 { + ffi_boundary!(-1, { + if config.is_null() || out.is_null() { + set_error("config or out-parameter pointer is null"); + return -1; + } + // SAFETY: config is a non-null `*const ThetaDataDxConfig` returned by `thetadatadx_config_*` and not yet freed; `&*` produces a shared reference valid for the call duration. + let config = unsafe { &*config }; + // SAFETY: out pointer checked non-null above; the FFI contract pins the storage for the call duration and forbids concurrent calls on the same handle. + unsafe { + *out = config.inner.historical.request_timeout_secs; + } + 0 + }) +} + // ── HistoricalClient ── /// Connect a historical client to `ThetaData` servers @@ -2841,6 +2885,41 @@ mod pool_sizing_tests { } } + #[test] + fn historical_request_timeout_secs_round_trips() { + let cfg = super::thetadatadx_config_production(); + assert!(!cfg.is_null()); + // SAFETY: handle just returned by thetadatadx_config_production. + unsafe { + // Default seeded at 300s by `HistoricalConfig::default()`. + let mut current: u64 = 0; + assert_eq!( + super::thetadatadx_config_get_request_timeout_secs(cfg, &mut current), + 0 + ); + assert_eq!(current, 300); + // Override. + super::thetadatadx_config_set_request_timeout_secs(cfg, 45); + assert_eq!((*cfg).inner.historical.request_timeout_secs, 45); + assert_eq!( + super::thetadatadx_config_get_request_timeout_secs(cfg, &mut current), + 0 + ); + assert_eq!(current, 45); + // Disable (no default deadline). + super::thetadatadx_config_set_request_timeout_secs(cfg, 0); + assert_eq!((*cfg).inner.historical.request_timeout_secs, 0); + // Null-pointer guards: setter is a no-op (matches the + // ffi_boundary `()` return); getter returns -1. + super::thetadatadx_config_set_request_timeout_secs(std::ptr::null_mut(), 4); + assert_eq!( + super::thetadatadx_config_get_request_timeout_secs(std::ptr::null(), &mut current), + -1 + ); + super::thetadatadx_config_free(cfg); + } + } + #[test] fn null_handle_is_safe() { // SAFETY: passing null to thetadatadx_config_set_* / thetadatadx_*_free is the diff --git a/sdks/cpp/include/thetadatadx.h b/sdks/cpp/include/thetadatadx.h index 3c41a40e1..cb1347cd9 100644 --- a/sdks/cpp/include/thetadatadx.h +++ b/sdks/cpp/include/thetadatadx.h @@ -1918,6 +1918,25 @@ void thetadatadx_config_set_warn_on_buffered_threshold_bytes(ThetaDataDxConfig* */ int32_t thetadatadx_config_get_warn_on_buffered_threshold_bytes(const ThetaDataDxConfig* config, size_t* out_n); +/** + * Set the default per-request deadline (seconds) for historical queries. + * + * Bounds every request that did not set its own deadline, so a + * live-but-silent stream resolves to a timeout instead of blocking + * forever. + * @param config Config handle to mutate; no-op when NULL. + * @param secs Deadline in seconds; 0 disables the default. Default 300. + */ +void thetadatadx_config_set_request_timeout_secs(ThetaDataDxConfig* config, uint64_t secs); + +/** + * Read the current historical request_timeout_secs setting. + * @param config Config handle to read. + * @param out Receives the configured seconds on success (0 = no default deadline). + * @return 0 on success, -1 if either pointer is null. + */ +int32_t thetadatadx_config_get_request_timeout_secs(const ThetaDataDxConfig* config, uint64_t* out); + /* ── HistoricalClient ── */ /** Connect a historical client to ThetaData servers. diff --git a/sdks/cpp/include/thetadatadx.hpp b/sdks/cpp/include/thetadatadx.hpp index 24a83f1de..4217e5955 100644 --- a/sdks/cpp/include/thetadatadx.hpp +++ b/sdks/cpp/include/thetadatadx.hpp @@ -1392,6 +1392,30 @@ class Config { return n; } + /** + * Set the default per-request deadline (seconds) for historical queries. + * + * Bounds every request that did not set its own deadline, so a + * live-but-silent stream resolves to a timeout instead of blocking + * forever. @p secs = 0 disables the default. Default is `300`. + */ + void set_request_timeout_secs(std::uint64_t secs) { + thetadatadx_config_set_request_timeout_secs(handle_.get(), secs); + } + + /** + * Read the current historical `request_timeout_secs` setting. + * + * Returns the configured seconds (`0` = no default deadline), or + * `0` on a null handle (matching the C ABI's `-1` failure mapping + * at the boundary). + */ + std::uint64_t get_request_timeout_secs() const { + std::uint64_t secs = 0; + thetadatadx_config_get_request_timeout_secs(handle_.get(), &secs); + return secs; + } + /** Get the raw handle. */ ThetaDataDxConfig* get() const { return handle_.get(); } diff --git a/sdks/cpp/tests/config_pool_sizing.cpp b/sdks/cpp/tests/config_pool_sizing.cpp index 9df5be7e5..3d2818673 100644 --- a/sdks/cpp/tests/config_pool_sizing.cpp +++ b/sdks/cpp/tests/config_pool_sizing.cpp @@ -26,6 +26,23 @@ TEST_CASE("Config concurrent_requests setter + getter round-trip", "[config][poo REQUIRE(cfg.get_concurrent_requests() == 32u); } +TEST_CASE("Config historical request_timeout_secs setter + getter round-trip", + "[config][pool_sizing][offline]") { + auto cfg = thetadatadx::Config::production(); + // The readback getter mirrors the Python `Config.request_timeout_secs` + // and TypeScript `requestTimeoutSecs` surfaces, so a value set + // through the C++ wrapper reads back through the same wrapper. + // Production seeds the 300s default per-request deadline. + REQUIRE(cfg.get_request_timeout_secs() == 300u); + cfg.set_request_timeout_secs(45); + REQUIRE(cfg.get_request_timeout_secs() == 45u); + cfg.set_request_timeout_secs(600); + REQUIRE(cfg.get_request_timeout_secs() == 600u); + // 0 disables the default deadline. + cfg.set_request_timeout_secs(0); + REQUIRE(cfg.get_request_timeout_secs() == 0u); +} + TEST_CASE("Config historical_host / historical_port setters + getters round-trip", "[config][historical][offline]") { // The historical endpoint overrides mirror the Python `Config.historical_host` / diff --git a/sdks/parity.toml b/sdks/parity.toml index 097865f45..50da90dca 100644 --- a/sdks/parity.toml +++ b/sdks/parity.toml @@ -286,6 +286,12 @@ python = true typescript = true cpp = true +[[class]] +name = "HistoricalConfig.request_timeout_secs" +python = true +typescript = true +cpp = true + # ─── ReconnectConfig cross-binding setters ───────────────────────────── # # Streaming reconnect policy + per-class attempt budgets + stable-window diff --git a/sdks/python/python/thetadatadx/__init__.pyi b/sdks/python/python/thetadatadx/__init__.pyi index 0a5d833c2..981cf5acd 100644 --- a/sdks/python/python/thetadatadx/__init__.pyi +++ b/sdks/python/python/thetadatadx/__init__.pyi @@ -107,6 +107,8 @@ class Config: """Maximum in-flight historical requests. ``0`` auto-detects the cap from the subscription tier; explicit values above the tier cap are clamped at connect time with a warning.""" warn_on_buffered_threshold_bytes: int """Byte ceiling above which a buffered (non-``.stream()``) historical response logs a warning pointing the caller at the streaming surface. ``0`` disables the warning; the default is ``100 * 1024 * 1024`` (100 MiB). The data is still delivered.""" + request_timeout_secs: int + """Default per-request deadline, in seconds, for historical queries. Bounds every request that did not set its own deadline, so a live-but-silent stream resolves to a timeout instead of blocking forever. ``0`` disables the default; the default is ``300`` (5 minutes).""" reconnect_policy: str """Active reconnect policy name: ``"auto"``, ``"manual"``, or ``"custom"`` (the last reported when a :attr:`reconnect_callback` is installed).""" reconnect_max_attempts: int diff --git a/sdks/python/src/lib.rs b/sdks/python/src/lib.rs index 80b83b7ca..40d51a140 100644 --- a/sdks/python/src/lib.rs +++ b/sdks/python/src/lib.rs @@ -1295,6 +1295,31 @@ impl Config { guard.historical.warn_on_buffered_threshold_bytes } + /// Set the default per-request deadline (in seconds) for historical + /// queries. Bounds every request that did not call + /// ``with_deadline(...)``, so a live-but-silent stream resolves to a + /// timeout instead of blocking forever. ``0`` disables the default + /// (no deadline unless the caller sets one). Default is ``300`` + /// (5 minutes). + /// + /// Examples + /// -------- + /// cfg = Config.production() + /// cfg.request_timeout_secs = 120 + #[setter] + fn set_request_timeout_secs(&self, secs: u64) { + let mut guard = self.inner.lock().unwrap_or_else(|e| e.into_inner()); + guard.historical.request_timeout_secs = secs; + } + + /// Current historical ``request_timeout_secs`` setting (``0`` = no + /// default deadline). + #[getter] + fn get_request_timeout_secs(&self) -> u64 { + let guard = self.inner.lock().unwrap_or_else(|e| e.into_inner()); + guard.historical.request_timeout_secs + } + fn __repr__(&self) -> String { let guard = self.inner.lock().unwrap_or_else(|e| e.into_inner()); format!( diff --git a/sdks/python/tests/test_config_pool_sizing.py b/sdks/python/tests/test_config_pool_sizing.py index 5ef907f06..532262809 100644 --- a/sdks/python/tests/test_config_pool_sizing.py +++ b/sdks/python/tests/test_config_pool_sizing.py @@ -43,3 +43,23 @@ def test_concurrent_requests_round_trips(): for n in (1, 2, 4, 8, 16): cfg.concurrent_requests = n assert cfg.concurrent_requests == n + + +# ─── request_timeout_secs ─────────────────────────────────────────── + + +def test_request_timeout_secs_defaults_to_300(): + """`request_timeout_secs` defaults to the 300s per-request deadline.""" + mod = _import_module() + cfg = mod.Config.production() + assert cfg.request_timeout_secs == 300 + + +def test_request_timeout_secs_round_trips(): + """`request_timeout_secs = N` round-trips through the binding; + ``0`` disables the default deadline.""" + mod = _import_module() + cfg = mod.Config.production() + for secs in (0, 1, 45, 120, 600): + cfg.request_timeout_secs = secs + assert cfg.request_timeout_secs == secs diff --git a/sdks/typescript/__tests__/config_pool_sizing.test.mjs b/sdks/typescript/__tests__/config_pool_sizing.test.mjs index 9a326e119..d8dc292f8 100644 --- a/sdks/typescript/__tests__/config_pool_sizing.test.mjs +++ b/sdks/typescript/__tests__/config_pool_sizing.test.mjs @@ -34,3 +34,18 @@ describe('Config.concurrentRequests', () => { } }); }); + +describe('Config.requestTimeoutSecs', () => { + it('defaults to 300n (5-minute per-request deadline)', () => { + const cfg = Config.production(); + assert.equal(cfg.requestTimeoutSecs, 300n); + }); + + it('round-trips through the setter; 0n disables the default', () => { + const cfg = Config.production(); + for (const secs of [0n, 1n, 45n, 120n, 600n]) { + cfg.setRequestTimeoutSecs(secs); + assert.equal(cfg.requestTimeoutSecs, secs); + } + }); +}); diff --git a/sdks/typescript/index.d.ts b/sdks/typescript/index.d.ts index 8255ce59e..65560e433 100644 --- a/sdks/typescript/index.d.ts +++ b/sdks/typescript/index.d.ts @@ -100,6 +100,20 @@ export declare class Config { * returned as a `BigInt`). */ get warnOnBufferedThresholdBytes(): bigint + /** + * Set the default per-request deadline (seconds) for historical + * queries. Bounds every request that did not set its own deadline, + * so a live-but-silent stream resolves to a timeout instead of + * blocking forever. `0n` disables the default. Default `300n` + * (5 minutes). Seconds are taken as a `BigInt` for parity with the + * other `*Secs` knobs. + */ + setRequestTimeoutSecs(secs: bigint): void + /** + * Current historical `request_timeout_secs` setting in seconds + * (default `300n`; `0n` = no default deadline). + */ + get requestTimeoutSecs(): bigint /** * Set the streaming reconnect policy. * diff --git a/sdks/typescript/src/config_class.rs b/sdks/typescript/src/config_class.rs index f89c85e3b..925433002 100644 --- a/sdks/typescript/src/config_class.rs +++ b/sdks/typescript/src/config_class.rs @@ -169,6 +169,39 @@ impl Config { )) } + /// Set the default per-request deadline (seconds) for historical + /// queries. Bounds every request that did not set its own deadline, + /// so a live-but-silent stream resolves to a timeout instead of + /// blocking forever. `0n` disables the default. Default `300n` + /// (5 minutes). Seconds are taken as a `BigInt` for parity with the + /// other `*Secs` knobs. + #[napi(js_name = "setRequestTimeoutSecs")] + pub fn set_request_timeout_secs( + &self, + secs: napi::bindgen_prelude::BigInt, + ) -> napi::Result<()> { + let value = bigint_to_u64("setRequestTimeoutSecs", &secs)?; + let mut guard = self + .inner + .lock() + .map_err(|_| napi::Error::from_reason("Config mutex poisoned"))?; + guard.historical.request_timeout_secs = value; + Ok(()) + } + + /// Current historical `request_timeout_secs` setting in seconds + /// (default `300n`; `0n` = no default deadline). + #[napi(getter, js_name = "requestTimeoutSecs")] + pub fn request_timeout_secs(&self) -> napi::Result { + let guard = self + .inner + .lock() + .map_err(|_| napi::Error::from_reason("Config mutex poisoned"))?; + Ok(napi::bindgen_prelude::BigInt::from( + guard.historical.request_timeout_secs, + )) + } + // ── Streaming reconnect knobs — parity with Python / C++ / FFI ─ /// Set the streaming reconnect policy.