From f977c165ba2b034526ffe5b5ba0f2e6b8d57106e Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 16 Jun 2026 10:09:48 +0200 Subject: [PATCH] fix(bindings): forward the full streaming config through the standalone clients The standalone Python and TypeScript StreamingClient builders only carried a subset of the streaming and reconnect tuning surface. Knobs like host selection, the watchdog and keepalive cadences, the reconnect backoff and replay pacing, and the newer wait strategy and consumer-core affinity were silently dropped, so tuning DirectConfig and then using a standalone client gave you builder defaults instead. Both bindings now snapshot the whole StreamingConfig and ReconnectConfig out of the user's Config and thread every field into StreamingClientBuilder, the same wholesale-snapshot the C ABI uses. That keeps the standalone connect path on the same ring-build and consumer-affinity code as the unified client, and a future config field cannot drift out of a hand-maintained subset. The C ABI builder was also missing the wait strategy and consumer-core knobs; it now forwards them too, so every client reaches the same connect surface. Each binding crate gets a regression test that sets every streaming and reconnect knob to a non-default value, snapshots the params, and asserts each one survives, so a forgotten field fails the build instead of dropping a user's tuning. Closes #802 Co-Authored-By: Claude Opus 4.8 --- CHANGELOG.md | 1 + docs-site/docs/changelog.md | 1 + ffi/src/streaming.rs | 7 ++ sdks/python/src/fpss_client.rs | 174 +++++++++++++++++++++++------ sdks/typescript/src/fpss_client.rs | 172 +++++++++++++++++++++++----- 5 files changed, 292 insertions(+), 63 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c1b83b319..835a28084 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -68,6 +68,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Streaming reconnect prefers the host that was last serving data: once a session survives the stable window the last-known-good address is pinned and tried first on the next reconnect, then the full configured host-selection policy runs. - The CLI's raw OHLC output (`--format json-raw` / `csv`) emits the `vwap` value in its own column. The raw value row was one field short of its header set, so `vwap` was dropped and `date`, `expiration`, `strike`, and `right` rendered under the wrong column names. The typed SDK surfaces and the CLI's presentation `json` were unaffected. - Python — the standalone `StreamingClient` streaming connect, reconnect, and subscribe / unsubscribe paths release the GIL across their blocking I/O (the TLS connect and handshake, and the per-subscription wire write), so other Python threads keep running while a connect or subscribe is in flight; the typed exception raised on failure is unchanged. +- The standalone Python and TypeScript streaming clients now forward the full streaming and reconnect config, so every tuning knob — including the wait strategy and consumer-core affinity, host selection, watchdog and keepalive cadences, and the reconnect backoff and replay pacing — is honored, matching the unified client and the C ABI. ### Security diff --git a/docs-site/docs/changelog.md b/docs-site/docs/changelog.md index c1b83b319..835a28084 100644 --- a/docs-site/docs/changelog.md +++ b/docs-site/docs/changelog.md @@ -68,6 +68,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Streaming reconnect prefers the host that was last serving data: once a session survives the stable window the last-known-good address is pinned and tried first on the next reconnect, then the full configured host-selection policy runs. - The CLI's raw OHLC output (`--format json-raw` / `csv`) emits the `vwap` value in its own column. The raw value row was one field short of its header set, so `vwap` was dropped and `date`, `expiration`, `strike`, and `right` rendered under the wrong column names. The typed SDK surfaces and the CLI's presentation `json` were unaffected. - Python — the standalone `StreamingClient` streaming connect, reconnect, and subscribe / unsubscribe paths release the GIL across their blocking I/O (the TLS connect and handshake, and the per-subscription wire write), so other Python threads keep running while a connect or subscribe is in flight; the typed exception raised on failure is unchanged. +- The standalone Python and TypeScript streaming clients now forward the full streaming and reconnect config, so every tuning knob — including the wait strategy and consumer-core affinity, host selection, watchdog and keepalive cadences, and the reconnect backoff and replay pacing — is honored, matching the unified client and the C ABI. ### Security diff --git a/ffi/src/streaming.rs b/ffi/src/streaming.rs index c13478d26..29aef45d8 100644 --- a/ffi/src/streaming.rs +++ b/ffi/src/streaming.rs @@ -193,6 +193,13 @@ fn streaming_builder( thetadatadx::fpss::StreamingClient::builder(¶ms.creds, ¶ms.streaming.hosts) .ring_size(params.streaming.ring_size) .flush_mode(params.streaming.flush_mode) + .wait_strategy(params.streaming.wait_strategy) + .wait_strategy_tuning( + params.streaming.wait_spin_iters, + params.streaming.wait_yield_iters, + params.streaming.wait_park_us, + ) + .consumer_cpu(params.streaming.consumer_cpu) .reconnect_policy(params.reconnect.policy.clone()) .reconnect_wait_ms(params.reconnect.wait_ms) .reconnect_wait_max_ms(params.reconnect.wait_max_ms) diff --git a/sdks/python/src/fpss_client.rs b/sdks/python/src/fpss_client.rs index 4106a01ff..cba36a7bd 100644 --- a/sdks/python/src/fpss_client.rs +++ b/sdks/python/src/fpss_client.rs @@ -55,49 +55,65 @@ use crate::{Config, Credentials}; /// Python-side mutations of the `Config` handle cannot retroactively /// change reconnect behaviour for an already-running session — the same /// snapshot semantics the FFI uses in -/// `ffi/src/streaming.rs::FpssConnectParams`. +/// `ffi/src/streaming.rs::StreamingConnectParams`. +/// +/// The whole [`StreamingConfig`] and [`ReconnectConfig`] are snapshotted +/// wholesale rather than copied field by field, so a new tuning knob +/// added to either config cannot drift out of the standalone connect +/// path the way a hand-maintained subset did. +/// +/// [`StreamingConfig`]: thetadatadx::config::StreamingConfig +/// [`ReconnectConfig`]: thetadatadx::config::ReconnectConfig struct FpssParams { creds: RustCredentials, - hosts: Vec<(String, u16)>, - ring_size: usize, - flush_mode: thetadatadx::config::StreamingFlushMode, - policy: thetadatadx::config::ReconnectPolicy, - wait_ms: u64, - wait_rate_limited_ms: u64, - derive_ohlcvc: bool, - connect_timeout_ms: u64, - read_timeout_ms: u64, - ping_interval_ms: u64, + streaming: thetadatadx::config::StreamingConfig, + reconnect: thetadatadx::config::ReconnectConfig, } impl FpssParams { fn from_config(creds: &RustCredentials, config: &DirectConfig) -> Self { Self { creds: creds.clone(), - hosts: config.streaming.hosts.clone(), - ring_size: config.streaming.ring_size, - flush_mode: config.streaming.flush_mode, - policy: config.reconnect.policy.clone(), - wait_ms: config.reconnect.wait_ms, - wait_rate_limited_ms: config.reconnect.wait_rate_limited_ms, - derive_ohlcvc: config.streaming.derive_ohlcvc, - connect_timeout_ms: config.streaming.connect_timeout_ms, - read_timeout_ms: config.streaming.timeout_ms, - ping_interval_ms: config.streaming.ping_interval_ms, + streaming: config.streaming.clone(), + reconnect: config.reconnect.clone(), } } + /// Thread every connection-side knob from the snapshot into a + /// [`fpss::StreamingClientBuilder`]. Kept in lockstep with the + /// unified client's connect path (`crates/thetadatadx/src/client.rs`) + /// and the C ABI (`ffi/src/streaming.rs::streaming_builder`) so the + /// standalone client honours the full streaming and reconnect surface. fn builder(&self) -> fpss::StreamingClientBuilder<'_> { - fpss::StreamingClientBuilder::new(&self.creds, &self.hosts) - .ring_size(self.ring_size) - .flush_mode(self.flush_mode) - .reconnect_policy(self.policy.clone()) - .reconnect_wait_ms(self.wait_ms) - .reconnect_wait_rate_limited_ms(self.wait_rate_limited_ms) - .derive_ohlcvc(self.derive_ohlcvc) - .connect_timeout_ms(self.connect_timeout_ms) - .read_timeout_ms(self.read_timeout_ms) - .ping_interval_ms(self.ping_interval_ms) + fpss::StreamingClientBuilder::new(&self.creds, &self.streaming.hosts) + .ring_size(self.streaming.ring_size) + .flush_mode(self.streaming.flush_mode) + .wait_strategy(self.streaming.wait_strategy) + .wait_strategy_tuning( + self.streaming.wait_spin_iters, + self.streaming.wait_yield_iters, + self.streaming.wait_park_us, + ) + .consumer_cpu(self.streaming.consumer_cpu) + .reconnect_policy(self.reconnect.policy.clone()) + .reconnect_wait_ms(self.reconnect.wait_ms) + .reconnect_wait_max_ms(self.reconnect.wait_max_ms) + .reconnect_wait_rate_limited_ms(self.reconnect.wait_rate_limited_ms) + .reconnect_wait_server_restart_ms(self.reconnect.wait_server_restart_ms) + .reconnect_jitter(self.reconnect.jitter) + .reconnect_replay_burst_size(self.reconnect.replay_burst_size) + .reconnect_replay_pace_ms(self.reconnect.replay_pace_ms) + .derive_ohlcvc(self.streaming.derive_ohlcvc) + .connect_timeout_ms(self.streaming.connect_timeout_ms) + .read_timeout_ms(self.streaming.timeout_ms) + .ping_interval_ms(self.streaming.ping_interval_ms) + .io_read_slice_ms(self.streaming.io_read_slice_ms) + .data_watchdog_ms(self.streaming.data_watchdog_ms) + .keepalive_idle_secs(self.streaming.keepalive_idle_secs) + .keepalive_interval_secs(self.streaming.keepalive_interval_secs) + .keepalive_retries(self.streaming.keepalive_retries) + .host_selection(self.streaming.host_selection) + .host_shuffle_seed(self.streaming.host_shuffle_seed) } } @@ -312,7 +328,7 @@ impl StreamingClient { } else { "none" }; - let hosts = self.params.hosts.len(); + let hosts = self.params.streaming.hosts.len(); format!("StreamingClient(streaming={streaming}, hosts={hosts})") } @@ -857,3 +873,97 @@ impl StreamingClient { ) } } + +#[cfg(test)] +mod tests { + use super::*; + use thetadatadx::config::{ + HostSelectionPolicy, JitterMode, ReconnectPolicy, StreamingFlushMode, StreamingWaitStrategy, + }; + + /// Anti-drift guard for the standalone connect path. + /// + /// `FpssParams` snapshots the whole `StreamingConfig` + `ReconnectConfig` + /// and `builder()` threads every field into the `StreamingClientBuilder`, + /// so the standalone Python `StreamingClient` honours the same streaming + /// and reconnect surface as the unified client and the C ABI. This test + /// sets every streaming and reconnect knob to a non-default value and + /// asserts each one survives the snapshot. A future field that + /// `from_config` forgets to carry makes this fail rather than silently + /// dropping a user's tuning. + #[test] + fn from_config_preserves_every_streaming_and_reconnect_knob() { + let creds = RustCredentials::new("user@example.com", "secret"); + let mut config = DirectConfig::production(); + + // Streaming: flip every knob away from its production default. + config.streaming.hosts = vec![("stream.example.com".to_owned(), 12345)]; + config.streaming.host_selection = HostSelectionPolicy::FixedOrder; + config.streaming.host_shuffle_seed = Some(0xABCD_1234); + config.streaming.timeout_ms = 111_111; + config.streaming.ring_size = 1 << 20; + config.streaming.ping_interval_ms = 22_222; + config.streaming.connect_timeout_ms = 33_333; + config.streaming.io_read_slice_ms = 44; + config.streaming.data_watchdog_ms = 55_555; + config.streaming.keepalive_idle_secs = 66; + config.streaming.keepalive_interval_secs = 77; + config.streaming.keepalive_retries = 8; + config.streaming.flush_mode = StreamingFlushMode::Immediate; + config.streaming.wait_strategy = StreamingWaitStrategy::Balanced; + config.streaming.wait_spin_iters = 123; + config.streaming.wait_yield_iters = 456; + config.streaming.wait_park_us = 789; + config.streaming.consumer_cpu = Some(3); + config.streaming.derive_ohlcvc = false; + + // Reconnect: flip every knob away from its production default. + config.reconnect.wait_ms = 1_010; + config.reconnect.wait_max_ms = 2_020; + config.reconnect.wait_rate_limited_ms = 3_030; + config.reconnect.wait_server_restart_ms = 4_040; + config.reconnect.jitter = JitterMode::None; + config.reconnect.replay_burst_size = 51; + config.reconnect.replay_pace_ms = 62; + config.reconnect.policy = ReconnectPolicy::Manual; + + let params = FpssParams::from_config(&creds, &config); + + let s = ¶ms.streaming; + assert_eq!(s.hosts, config.streaming.hosts); + assert_eq!(s.host_selection, HostSelectionPolicy::FixedOrder); + assert_eq!(s.host_shuffle_seed, Some(0xABCD_1234)); + assert_eq!(s.timeout_ms, 111_111); + assert_eq!(s.ring_size, 1 << 20); + assert_eq!(s.ping_interval_ms, 22_222); + assert_eq!(s.connect_timeout_ms, 33_333); + assert_eq!(s.io_read_slice_ms, 44); + assert_eq!(s.data_watchdog_ms, 55_555); + assert_eq!(s.keepalive_idle_secs, 66); + assert_eq!(s.keepalive_interval_secs, 77); + assert_eq!(s.keepalive_retries, 8); + assert_eq!(s.flush_mode, StreamingFlushMode::Immediate); + assert_eq!(s.wait_strategy, StreamingWaitStrategy::Balanced); + assert_eq!(s.wait_spin_iters, 123); + assert_eq!(s.wait_yield_iters, 456); + assert_eq!(s.wait_park_us, 789); + assert_eq!(s.consumer_cpu, Some(3)); + assert!(!s.derive_ohlcvc); + + let r = ¶ms.reconnect; + assert_eq!(r.wait_ms, 1_010); + assert_eq!(r.wait_max_ms, 2_020); + assert_eq!(r.wait_rate_limited_ms, 3_030); + assert_eq!(r.wait_server_restart_ms, 4_040); + assert_eq!(r.jitter, JitterMode::None); + assert_eq!(r.replay_burst_size, 51); + assert_eq!(r.replay_pace_ms, 62); + assert!( + matches!(r.policy, ReconnectPolicy::Manual), + "reconnect policy must survive the snapshot" + ); + + // The snapshot must build without panicking with every knob set. + let _ = params.builder(); + } +} diff --git a/sdks/typescript/src/fpss_client.rs b/sdks/typescript/src/fpss_client.rs index 7d95afccf..b9df9dec8 100644 --- a/sdks/typescript/src/fpss_client.rs +++ b/sdks/typescript/src/fpss_client.rs @@ -64,49 +64,65 @@ use crate::{ /// mutations of the `Config` handle cannot retroactively change reconnect /// behaviour for an already-running session — the same snapshot semantics /// the Python binding (`FpssParams`) and the FFI -/// (`ffi/src/streaming.rs::FpssConnectParams`) use. +/// (`ffi/src/streaming.rs::StreamingConnectParams`) use. +/// +/// The whole [`StreamingConfig`] and [`ReconnectConfig`] are snapshotted +/// wholesale rather than copied field by field, so a new tuning knob +/// added to either config cannot drift out of the standalone connect +/// path the way a hand-maintained subset did. +/// +/// [`StreamingConfig`]: thetadatadx::config::StreamingConfig +/// [`ReconnectConfig`]: thetadatadx::config::ReconnectConfig struct FpssParams { creds: RustCredentials, - hosts: Vec<(String, u16)>, - ring_size: usize, - flush_mode: thetadatadx::config::StreamingFlushMode, - policy: thetadatadx::config::ReconnectPolicy, - wait_ms: u64, - wait_rate_limited_ms: u64, - derive_ohlcvc: bool, - connect_timeout_ms: u64, - read_timeout_ms: u64, - ping_interval_ms: u64, + streaming: thetadatadx::config::StreamingConfig, + reconnect: thetadatadx::config::ReconnectConfig, } impl FpssParams { fn from_config(creds: &RustCredentials, config: &DirectConfig) -> Self { Self { creds: creds.clone(), - hosts: config.streaming.hosts.clone(), - ring_size: config.streaming.ring_size, - flush_mode: config.streaming.flush_mode, - policy: config.reconnect.policy.clone(), - wait_ms: config.reconnect.wait_ms, - wait_rate_limited_ms: config.reconnect.wait_rate_limited_ms, - derive_ohlcvc: config.streaming.derive_ohlcvc, - connect_timeout_ms: config.streaming.connect_timeout_ms, - read_timeout_ms: config.streaming.timeout_ms, - ping_interval_ms: config.streaming.ping_interval_ms, + streaming: config.streaming.clone(), + reconnect: config.reconnect.clone(), } } + /// Thread every connection-side knob from the snapshot into a + /// [`fpss::StreamingClientBuilder`]. Kept in lockstep with the + /// unified client's connect path (`crates/thetadatadx/src/client.rs`) + /// and the C ABI (`ffi/src/streaming.rs::streaming_builder`) so the + /// standalone client honours the full streaming and reconnect surface. fn builder(&self) -> fpss::StreamingClientBuilder<'_> { - fpss::StreamingClientBuilder::new(&self.creds, &self.hosts) - .ring_size(self.ring_size) - .flush_mode(self.flush_mode) - .reconnect_policy(self.policy.clone()) - .reconnect_wait_ms(self.wait_ms) - .reconnect_wait_rate_limited_ms(self.wait_rate_limited_ms) - .derive_ohlcvc(self.derive_ohlcvc) - .connect_timeout_ms(self.connect_timeout_ms) - .read_timeout_ms(self.read_timeout_ms) - .ping_interval_ms(self.ping_interval_ms) + fpss::StreamingClientBuilder::new(&self.creds, &self.streaming.hosts) + .ring_size(self.streaming.ring_size) + .flush_mode(self.streaming.flush_mode) + .wait_strategy(self.streaming.wait_strategy) + .wait_strategy_tuning( + self.streaming.wait_spin_iters, + self.streaming.wait_yield_iters, + self.streaming.wait_park_us, + ) + .consumer_cpu(self.streaming.consumer_cpu) + .reconnect_policy(self.reconnect.policy.clone()) + .reconnect_wait_ms(self.reconnect.wait_ms) + .reconnect_wait_max_ms(self.reconnect.wait_max_ms) + .reconnect_wait_rate_limited_ms(self.reconnect.wait_rate_limited_ms) + .reconnect_wait_server_restart_ms(self.reconnect.wait_server_restart_ms) + .reconnect_jitter(self.reconnect.jitter) + .reconnect_replay_burst_size(self.reconnect.replay_burst_size) + .reconnect_replay_pace_ms(self.reconnect.replay_pace_ms) + .derive_ohlcvc(self.streaming.derive_ohlcvc) + .connect_timeout_ms(self.streaming.connect_timeout_ms) + .read_timeout_ms(self.streaming.timeout_ms) + .ping_interval_ms(self.streaming.ping_interval_ms) + .io_read_slice_ms(self.streaming.io_read_slice_ms) + .data_watchdog_ms(self.streaming.data_watchdog_ms) + .keepalive_idle_secs(self.streaming.keepalive_idle_secs) + .keepalive_interval_secs(self.streaming.keepalive_interval_secs) + .keepalive_retries(self.streaming.keepalive_retries) + .host_selection(self.streaming.host_selection) + .host_shuffle_seed(self.streaming.host_shuffle_seed) } } @@ -725,3 +741,97 @@ impl StreamingClient { } } } + +#[cfg(test)] +mod tests { + use super::*; + use thetadatadx::config::{ + HostSelectionPolicy, JitterMode, ReconnectPolicy, StreamingFlushMode, StreamingWaitStrategy, + }; + + /// Anti-drift guard for the standalone connect path. + /// + /// `FpssParams` snapshots the whole `StreamingConfig` + `ReconnectConfig` + /// and `builder()` threads every field into the `StreamingClientBuilder`, + /// so the standalone TypeScript `StreamingClient` honours the same + /// streaming and reconnect surface as the unified client and the C ABI. + /// This test sets every streaming and reconnect knob to a non-default + /// value and asserts each one survives the snapshot. A future field that + /// `from_config` forgets to carry makes this fail rather than silently + /// dropping a user's tuning. + #[test] + fn from_config_preserves_every_streaming_and_reconnect_knob() { + let creds = RustCredentials::new("user@example.com", "secret"); + let mut config = DirectConfig::production(); + + // Streaming: flip every knob away from its production default. + config.streaming.hosts = vec![("stream.example.com".to_owned(), 12345)]; + config.streaming.host_selection = HostSelectionPolicy::FixedOrder; + config.streaming.host_shuffle_seed = Some(0xABCD_1234); + config.streaming.timeout_ms = 111_111; + config.streaming.ring_size = 1 << 20; + config.streaming.ping_interval_ms = 22_222; + config.streaming.connect_timeout_ms = 33_333; + config.streaming.io_read_slice_ms = 44; + config.streaming.data_watchdog_ms = 55_555; + config.streaming.keepalive_idle_secs = 66; + config.streaming.keepalive_interval_secs = 77; + config.streaming.keepalive_retries = 8; + config.streaming.flush_mode = StreamingFlushMode::Immediate; + config.streaming.wait_strategy = StreamingWaitStrategy::Balanced; + config.streaming.wait_spin_iters = 123; + config.streaming.wait_yield_iters = 456; + config.streaming.wait_park_us = 789; + config.streaming.consumer_cpu = Some(3); + config.streaming.derive_ohlcvc = false; + + // Reconnect: flip every knob away from its production default. + config.reconnect.wait_ms = 1_010; + config.reconnect.wait_max_ms = 2_020; + config.reconnect.wait_rate_limited_ms = 3_030; + config.reconnect.wait_server_restart_ms = 4_040; + config.reconnect.jitter = JitterMode::None; + config.reconnect.replay_burst_size = 51; + config.reconnect.replay_pace_ms = 62; + config.reconnect.policy = ReconnectPolicy::Manual; + + let params = FpssParams::from_config(&creds, &config); + + let s = ¶ms.streaming; + assert_eq!(s.hosts, config.streaming.hosts); + assert_eq!(s.host_selection, HostSelectionPolicy::FixedOrder); + assert_eq!(s.host_shuffle_seed, Some(0xABCD_1234)); + assert_eq!(s.timeout_ms, 111_111); + assert_eq!(s.ring_size, 1 << 20); + assert_eq!(s.ping_interval_ms, 22_222); + assert_eq!(s.connect_timeout_ms, 33_333); + assert_eq!(s.io_read_slice_ms, 44); + assert_eq!(s.data_watchdog_ms, 55_555); + assert_eq!(s.keepalive_idle_secs, 66); + assert_eq!(s.keepalive_interval_secs, 77); + assert_eq!(s.keepalive_retries, 8); + assert_eq!(s.flush_mode, StreamingFlushMode::Immediate); + assert_eq!(s.wait_strategy, StreamingWaitStrategy::Balanced); + assert_eq!(s.wait_spin_iters, 123); + assert_eq!(s.wait_yield_iters, 456); + assert_eq!(s.wait_park_us, 789); + assert_eq!(s.consumer_cpu, Some(3)); + assert!(!s.derive_ohlcvc); + + let r = ¶ms.reconnect; + assert_eq!(r.wait_ms, 1_010); + assert_eq!(r.wait_max_ms, 2_020); + assert_eq!(r.wait_rate_limited_ms, 3_030); + assert_eq!(r.wait_server_restart_ms, 4_040); + assert_eq!(r.jitter, JitterMode::None); + assert_eq!(r.replay_burst_size, 51); + assert_eq!(r.replay_pace_ms, 62); + assert!( + matches!(r.policy, ReconnectPolicy::Manual), + "reconnect policy must survive the snapshot" + ); + + // The snapshot must build without panicking with every knob set. + let _ = params.builder(); + } +}