diff --git a/CHANGELOG.md b/CHANGELOG.md index 75447079..4a5cd706 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **`flush_mode` streaming write-flush knob.** The `flush_mode` setting is removed from every binding (Rust `StreamingConfig::flush_mode`, Python `Config.flush_mode`, TypeScript `Config.flushMode` / `setFlushMode`, C++ `set_flush_mode` / `get_flush_mode`, C ABI `thetadatadx_config_set_flush_mode` / `_get_flush_mode`). Outbound streaming writes now always coalesce and flush on the ping heartbeat, so a subscription burst leaves as fewer, larger packets — the terminal's own behavior — with received-data latency unaffected as before. The `"immediate"` per-frame-flush mode existed only to defeat that server-friendly coalescing and is gone. This is a breaking change to the configuration surface. +- **`host_selection` / `host_shuffle_seed` streaming host-ordering knobs.** Removed from every binding (Rust `StreamingConfig::host_selection` / `host_shuffle_seed`, Python `Config.streaming_host_selection` / `streaming_host_shuffle_seed`, TypeScript `Config.streamingHostSelection` / `setStreamingHostSelection` and the shuffle-seed pair, C++ `set_streaming_host_selection` / `get_streaming_host_selection` and the shuffle-seed pair, C ABI `thetadatadx_config_*_streaming_host_selection` / `_host_shuffle_seed`), along with the `HostSelectionPolicy` enum. The client now cycles the declared host list left to right — the terminal's own behavior — and a reconnect tries the last-known-good host first. The per-client fault-domain shuffle and its seed existed only in the SDK, with no terminal counterpart. This is a breaking change to the configuration surface. + ## [0.1.1] - 2026-07-07 ### Added diff --git a/docs-site/docs/changelog.md b/docs-site/docs/changelog.md index 75447079..4a5cd706 100644 --- a/docs-site/docs/changelog.md +++ b/docs-site/docs/changelog.md @@ -11,6 +11,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **`flush_mode` streaming write-flush knob.** The `flush_mode` setting is removed from every binding (Rust `StreamingConfig::flush_mode`, Python `Config.flush_mode`, TypeScript `Config.flushMode` / `setFlushMode`, C++ `set_flush_mode` / `get_flush_mode`, C ABI `thetadatadx_config_set_flush_mode` / `_get_flush_mode`). Outbound streaming writes now always coalesce and flush on the ping heartbeat, so a subscription burst leaves as fewer, larger packets — the terminal's own behavior — with received-data latency unaffected as before. The `"immediate"` per-frame-flush mode existed only to defeat that server-friendly coalescing and is gone. This is a breaking change to the configuration surface. +- **`host_selection` / `host_shuffle_seed` streaming host-ordering knobs.** Removed from every binding (Rust `StreamingConfig::host_selection` / `host_shuffle_seed`, Python `Config.streaming_host_selection` / `streaming_host_shuffle_seed`, TypeScript `Config.streamingHostSelection` / `setStreamingHostSelection` and the shuffle-seed pair, C++ `set_streaming_host_selection` / `get_streaming_host_selection` and the shuffle-seed pair, C ABI `thetadatadx_config_*_streaming_host_selection` / `_host_shuffle_seed`), along with the `HostSelectionPolicy` enum. The client now cycles the declared host list left to right — the terminal's own behavior — and a reconnect tries the last-known-good host first. The per-client fault-domain shuffle and its seed existed only in the SDK, with no terminal counterpart. This is a breaking change to the configuration surface. + ## [0.1.1] - 2026-07-07 ### Added diff --git a/parity.toml b/parity.toml index 6a824775..34dede1e 100644 --- a/parity.toml +++ b/parity.toml @@ -659,18 +659,6 @@ python = true typescript = true cpp = true -[[class]] -name = "StreamingConfig.streaming_host_selection" -python = true -typescript = true -cpp = true - -[[class]] -name = "StreamingConfig.streaming_host_shuffle_seed" -python = true -typescript = true -cpp = true - # `StreamingConfig.consumer_cpu` optional consumer-thread CPU pin, # exposed on every binding (Python `Config.consumer_cpu` `Optional[int]`, # TypeScript `consumerCpu` / `setConsumerCpu` `number | null`, C++ @@ -1955,15 +1943,14 @@ cpp = true returns = "String" python_pyi_returns = 'Literal["PROD", "DEV"]' -# The reconnect-jitter and streaming-host-selection readbacks. Both are -# value-constrained enums exposed as readback getters on every binding: Python -# (`Config.reconnect_jitter` / `Config.streaming_host_selection` properties), -# TypeScript (`get reconnectJitter()` / `get streamingHostSelection()`), C++ -# (`Config::get_reconnect_jitter()` / `Config::get_streaming_host_selection()`, -# each returning the enum's `int` discriminant). The `.pyi` constrains each to -# its exact value set, pinned via `python_pyi_returns` so a value-set drift -# fails — without these getter rows the constrained knobs were presence-tracked -# only (the config-field roster), never type-checked on the stub. +# The reconnect-jitter readback. A value-constrained enum exposed as a +# readback getter on every binding: Python (`Config.reconnect_jitter` +# property), TypeScript (`get reconnectJitter()`), C++ +# (`Config::get_reconnect_jitter()`, returning the enum's `int` +# discriminant). The `.pyi` constrains it to its exact value set, pinned +# via `python_pyi_returns` so a value-set drift fails — without this getter +# row the constrained knob was presence-tracked only (the config-field +# roster), never type-checked on the stub. [[method]] class = "Config" name = "reconnectJitter" @@ -1975,17 +1962,6 @@ returns = "String" cpp_returns = "i32" python_pyi_returns = 'Literal["full", "equal", "decorrelated", "none"]' -[[method]] -class = "Config" -name = "streamingHostSelection" -python = true -typescript = true -cpp = true -[method.signature] -returns = "String" -cpp_returns = "i32" -python_pyi_returns = 'Literal["shuffled", "fixed_order"]' - # Custom reconnect-policy callback registration. A field-shaped getter # is meaningless for a callback, so the registration is tracked as a # method row (Python `config.reconnect_callback = fn` setter, TS diff --git a/scripts/ci/check_binding_parity.py b/scripts/ci/check_binding_parity.py index c8d5f8de..602d8ec0 100644 --- a/scripts/ci/check_binding_parity.py +++ b/scripts/ci/check_binding_parity.py @@ -1861,8 +1861,6 @@ def _collect_rust_pub_fields(config_dir: pathlib.Path) -> dict[str, set[str]]: ("StreamingConfig", "keepalive_idle_secs"): "streaming_keepalive_idle_secs", ("StreamingConfig", "keepalive_interval_secs"): "streaming_keepalive_interval_secs", ("StreamingConfig", "keepalive_retries"): "streaming_keepalive_retries", - ("StreamingConfig", "host_selection"): "streaming_host_selection", - ("StreamingConfig", "host_shuffle_seed"): "streaming_host_shuffle_seed", } diff --git a/thetadatadx-cpp/include/config_accessors.hpp.inc b/thetadatadx-cpp/include/config_accessors.hpp.inc index 896437f7..bb54f544 100644 --- a/thetadatadx-cpp/include/config_accessors.hpp.inc +++ b/thetadatadx-cpp/include/config_accessors.hpp.inc @@ -515,25 +515,6 @@ return out; } - /// Set the streaming host-selection policy: 0=Shuffled (default), - /// 1=FixedOrder. Throws @c thetadatadx::InvalidParameterError on an - /// out-of-domain policy (and @c thetadatadx::ThetaDataError on a null - /// handle), routing through the typed leaf the FFI error code - /// selects. - void set_streaming_host_selection(int policy) { - if (thetadatadx_config_set_streaming_host_selection(handle_.get(), policy) != 0) { - detail::throw_last_ffi_error(); - } - } - - /// Current streaming host-selection policy (same encoding as the - /// setter). - int get_streaming_host_selection() const { - int32_t out{}; - thetadatadx_config_get_streaming_host_selection(handle_.get(), &out); - return out; - } - /// Set the Prometheus exporter port. Pass @c std::nullopt to leave /// the exporter disabled (the @c None default); pass an explicit /// @c std::uint16_t to bind an HTTP listener on @c 0.0.0.0: @@ -563,26 +544,3 @@ } return has_value ? std::optional{port} : std::nullopt; } - - /// Set the streaming host-shuffle seed using the (has_value, seed) - /// shape. has_value=false derives a fresh per-client seed; - /// has_value=true makes the shuffled order deterministic. - void set_streaming_host_shuffle_seed(std::optional seed) { - const bool has_value = seed.has_value(); - const std::uint64_t arg = seed.value_or(0); - if (thetadatadx_config_set_streaming_host_shuffle_seed(handle_.get(), has_value, arg) != 0) { - detail::throw_last_ffi_error(); - } - } - - /// Read the streaming host-shuffle seed back. Returns @c std::nullopt for - /// the per-client-entropy sentinel (no pinned seed); returns the - /// wrapped seed when the shuffled order is deterministic. - std::optional get_streaming_host_shuffle_seed() const { - bool has_value = false; - std::uint64_t seed = 0; - if (thetadatadx_config_get_streaming_host_shuffle_seed(handle_.get(), &has_value, &seed) != 0) { - detail::throw_last_ffi_error(); - } - return has_value ? std::optional{seed} : std::nullopt; - } diff --git a/thetadatadx-cpp/include/thetadatadx.h b/thetadatadx-cpp/include/thetadatadx.h index efce7a49..d413f823 100644 --- a/thetadatadx-cpp/include/thetadatadx.h +++ b/thetadatadx-cpp/include/thetadatadx.h @@ -1595,50 +1595,6 @@ void thetadatadx_config_set_streaming_ring_size(ThetaDataDxConfig* config, size_ */ int32_t thetadatadx_config_get_streaming_ring_size(const ThetaDataDxConfig* config, size_t* out); -/** - * Set the streaming host-selection policy. - * policy=0: Shuffled (default) -- fault-domain-aware per-client - * shuffle; a fleet spreads across hosts and consecutive - * failover attempts cross physical machines. - * policy=1: FixedOrder -- use the declared host order verbatim. - * @param config Config handle to mutate. - * @param policy Host-selection policy selector (0 = Shuffled, 1 = FixedOrder). - * @return 0 on success, -1 on an invalid policy or null config. - */ -int32_t thetadatadx_config_set_streaming_host_selection(ThetaDataDxConfig* config, int32_t policy); - -/** - * Read the configured streaming host-selection policy. Same encoding as - * thetadatadx_config_set_streaming_host_selection. - * @param config Config handle to read. - * @param out_policy Receives the host-selection policy on success. - * @return 0 on success, -1 if either pointer is null. - */ -int32_t thetadatadx_config_get_streaming_host_selection(const ThetaDataDxConfig* config, int32_t* out_policy); - -/** - * Set the streaming host-shuffle seed using the (has_value, seed) widened - * shape. Ignored under the FixedOrder policy. - * @param config Config handle to mutate. - * @param has_value false (default) derives a fresh per-client seed so a - * fleet shuffles independently; true makes the shuffled - * order deterministic, useful for fleet sharding and tests. - * @param seed The deterministic seed, honoured only when has_value is true. - * @return 0 on success, -1 if config is null. - */ -int32_t thetadatadx_config_set_streaming_host_shuffle_seed(ThetaDataDxConfig* config, bool has_value, uint64_t seed); - -/** - * Read the current streaming host-shuffle seed. - * @param config Config handle to read. - * @param out_has_value Receives false for the per-client-entropy sentinel, - * true when an explicit seed is set. - * @param out_seed Receives the seed when out_has_value is true. - * @return 0 on success, -1 if any pointer is null. - */ -int32_t thetadatadx_config_get_streaming_host_shuffle_seed(const ThetaDataDxConfig* config, bool* out_has_value, - uint64_t* out_seed); - /** * Set the wall-clock envelope (seconds) for one market-data-channel * retry sequence, measured from the first attempt. 0 disables the diff --git a/thetadatadx-cpp/tests/error_taxonomy.cpp b/thetadatadx-cpp/tests/error_taxonomy.cpp index ba0e7dba..47a70d93 100644 --- a/thetadatadx-cpp/tests/error_taxonomy.cpp +++ b/thetadatadx-cpp/tests/error_taxonomy.cpp @@ -211,8 +211,6 @@ TEST_CASE("config enum setters reject an out-of-domain value with InvalidParamet REQUIRE_NOTHROW(cfg.set_reconnect_jitter(2)); REQUIRE_THROWS_AS(cfg.set_reconnect_jitter(9), thetadatadx::InvalidParameterError); - REQUIRE_NOTHROW(cfg.set_streaming_host_selection(1)); - REQUIRE_THROWS_AS(cfg.set_streaming_host_selection(5), thetadatadx::InvalidParameterError); } TEST_CASE("sequence converters reject out-of-wire-range inputs with InvalidParameterError", diff --git a/thetadatadx-ffi/src/auth.rs b/thetadatadx-ffi/src/auth.rs index 22d687ef..dfbd65e5 100644 --- a/thetadatadx-ffi/src/auth.rs +++ b/thetadatadx-ffi/src/auth.rs @@ -2216,84 +2216,6 @@ mod resilience_knob_tests { } } - #[test] - fn fpss_host_selection_and_seed_round_trip() { - let cfg = super::thetadatadx_config_production(); - // SAFETY: handle just returned by thetadatadx_config_production. - unsafe { - let mut policy: i32 = -1; - assert_eq!( - super::thetadatadx_config_get_streaming_host_selection(cfg, &mut policy), - 0 - ); - assert_eq!(policy, 0, "default host selection is Shuffled"); - assert_eq!( - super::thetadatadx_config_set_streaming_host_selection(cfg, 1), - 0 - ); - assert_eq!( - super::thetadatadx_config_get_streaming_host_selection(cfg, &mut policy), - 0 - ); - assert_eq!(policy, 1); - assert_eq!( - super::thetadatadx_config_set_streaming_host_selection(cfg, 5), - -1, - "invalid policy rejected" - ); - assert_eq!( - crate::error::thetadatadx_last_error_code(), - crate::error::THETADATADX_ERR_INVALID_PARAMETER, - "a rejected enum value surfaces the typed invalid-parameter class" - ); - - let mut has_value = true; - let mut seed: u64 = 7; - assert_eq!( - super::thetadatadx_config_get_streaming_host_shuffle_seed( - cfg, - &mut has_value, - &mut seed - ), - 0 - ); - assert!( - !has_value, - "default seed is the per-client-entropy sentinel" - ); - assert_eq!(seed, 0); - assert_eq!( - super::thetadatadx_config_set_streaming_host_shuffle_seed(cfg, true, 42), - 0 - ); - assert_eq!( - super::thetadatadx_config_get_streaming_host_shuffle_seed( - cfg, - &mut has_value, - &mut seed - ), - 0 - ); - assert!(has_value); - assert_eq!(seed, 42); - assert_eq!( - super::thetadatadx_config_set_streaming_host_shuffle_seed(cfg, false, 0), - 0 - ); - assert_eq!( - super::thetadatadx_config_get_streaming_host_shuffle_seed( - cfg, - &mut has_value, - &mut seed - ), - 0 - ); - assert!(!has_value, "explicit None restores the sentinel"); - - super::thetadatadx_config_free(cfg); - } - } - #[test] fn retry_envelope_and_flatfiles_jitter_round_trip() { let cfg = super::thetadatadx_config_production(); diff --git a/thetadatadx-ffi/src/config_accessors.rs b/thetadatadx-ffi/src/config_accessors.rs index 6e09ec26..19063d88 100644 --- a/thetadatadx-ffi/src/config_accessors.rs +++ b/thetadatadx-ffi/src/config_accessors.rs @@ -1533,76 +1533,6 @@ pub unsafe extern "C" fn thetadatadx_config_get_reconnect_jitter( }) } -/// Set the streaming host-selection policy. -/// -/// - `policy = 0`: Shuffled (default) — fault-domain-aware per-client -/// shuffle; a fleet spreads across hosts and consecutive failover -/// attempts cross physical machines. -/// - `policy = 1`: FixedOrder — use the declared host order verbatim. -/// -/// Returns `0` on success. Returns `-1` and sets `thetadatadx_last_error` -/// when `policy` is outside the documented `{0, 1}` set or `config` -/// is null. A rejected `policy` value carries -/// `thetadatadx_last_error_code = THETADATADX_ERR_INVALID_PARAMETER` so an out-of-domain -/// enum int surfaces the same typed class across every binding. -#[no_mangle] -pub unsafe extern "C" fn thetadatadx_config_set_streaming_host_selection( - config: *mut ThetaDataDxConfig, - policy: i32, -) -> i32 { - ffi_boundary!(-1, { - if config.is_null() { - set_error("thetadatadx_config_set_streaming_host_selection: config handle is null"); - return -1; - } - let value = match policy { - 0 => thetadatadx::HostSelectionPolicy::Shuffled, - 1 => thetadatadx::HostSelectionPolicy::FixedOrder, - other => { - crate::error::set_error_with_code( - &format!( - "thetadatadx_config_set_streaming_host_selection: invalid policy {other}; expected 0 (Shuffled) or 1 (FixedOrder)" - ), - crate::error::THETADATADX_ERR_INVALID_PARAMETER, - ); - return -1; - } - }; - // SAFETY: config is a non-null pointer returned by `thetadatadx_config_*` and not yet freed; `&mut *` produces a unique reference valid for the call duration because the caller owns the Box and the FFI contract forbids concurrent calls on the same handle. - let config = unsafe { &mut *config }; - config.inner.streaming.host_selection = value; - 0 - }) -} - -/// Read the configured streaming host-selection policy. Same encoding as -/// `thetadatadx_config_set_streaming_host_selection`. Returns `0` on success, `-1` -/// if either pointer is null. -#[no_mangle] -pub unsafe extern "C" fn thetadatadx_config_get_streaming_host_selection( - config: *const ThetaDataDxConfig, - out_policy: *mut i32, -) -> i32 { - ffi_boundary!(-1, { - if config.is_null() || out_policy.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 }; - let value = match config.inner.streaming.host_selection { - thetadatadx::HostSelectionPolicy::Shuffled => 0, - thetadatadx::HostSelectionPolicy::FixedOrder => 1, - _ => 1, - }; - // 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_policy = value; - } - 0 - }) -} - /// Set the Prometheus exporter port on a config handle. /// /// * `has_value = false` encodes `None`: the exporter stays disabled @@ -1665,66 +1595,3 @@ pub unsafe extern "C" fn thetadatadx_config_get_metrics_port( }) } -/// Set the streaming host-shuffle seed using the `(has_value, seed)` -/// widened ABI shape that preserves the `None` sentinel across the C -/// boundary. -/// -/// * `has_value = false` → `None` (default): every client derives a -/// fresh per-instance seed, so a fleet shuffles independently. -/// `seed` is ignored. -/// * `has_value = true` → `Some(seed)`: the shuffled order becomes -/// deterministic — useful for fleet sharding and tests. -/// -/// Ignored under the `FixedOrder` host-selection policy. Returns `0` -/// on success, `-1` if `config` is null. -#[no_mangle] -pub unsafe extern "C" fn thetadatadx_config_set_streaming_host_shuffle_seed( - config: *mut ThetaDataDxConfig, - has_value: bool, - seed: u64, -) -> i32 { - ffi_boundary!(-1, { - if config.is_null() { - set_error("config handle is null"); - return -1; - } - // SAFETY: config is a non-null pointer returned by `thetadatadx_config_*` and not yet freed; `&mut *` produces a unique reference valid for the call duration because the caller owns the Box and the FFI contract forbids concurrent calls on the same handle. - let config = unsafe { &mut *config }; - config.inner.streaming.host_shuffle_seed = if has_value { Some(seed) } else { None }; - 0 - }) -} - -/// Read the current streaming host-shuffle seed. Same `(has_value, seed)` -/// ABI as `thetadatadx_config_set_streaming_host_shuffle_seed`: -/// -/// * `*out_has_value = false` → `None` (per-client entropy). `*out_seed` is left `0`. -/// * `*out_has_value = true` → `Some(*out_seed)`. -/// -/// Returns `0` on success, `-1` if any pointer is null. -#[no_mangle] -pub unsafe extern "C" fn thetadatadx_config_get_streaming_host_shuffle_seed( - config: *const ThetaDataDxConfig, - out_has_value: *mut bool, - out_seed: *mut u64, -) -> i32 { - ffi_boundary!(-1, { - if config.is_null() || out_has_value.is_null() || out_seed.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 }; - let (has_value, value) = match config.inner.streaming.host_shuffle_seed { - Some(v) => (true, v), - None => (false, 0), - }; - // SAFETY: out_has_value / out_seed null-checked above; caller pins the storage they point at for the call duration. - unsafe { - *out_has_value = has_value; - *out_seed = value; - } - 0 - }) -} - diff --git a/thetadatadx-ffi/src/streaming.rs b/thetadatadx-ffi/src/streaming.rs index 78805227..4175fb67 100644 --- a/thetadatadx-ffi/src/streaming.rs +++ b/thetadatadx-ffi/src/streaming.rs @@ -212,8 +212,7 @@ pub struct ThetaDataDxStreamHandle { struct StreamingConnectParams { creds: thetadatadx::Credentials, /// Snapshot of `DirectConfig.streaming` at handle-construction time — - /// hosts, ring size, timeouts, keepalive schedule, host-selection - /// policy. + /// hosts, ring size, timeouts, keepalive schedule. streaming: thetadatadx::config::StreamingConfig, /// Snapshot of `DirectConfig.reconnect` at handle-construction /// time — policy, per-class cadences, jitter, replay pacing. @@ -247,8 +246,6 @@ fn streaming_builder( .keepalive_idle_secs(params.streaming.keepalive_idle_secs) .keepalive_interval_secs(params.streaming.keepalive_interval_secs) .keepalive_retries(params.streaming.keepalive_retries) - .host_selection(params.streaming.host_selection) - .host_shuffle_seed(params.streaming.host_shuffle_seed) } // ═══════════════════════════════════════════════════════════════════════ diff --git a/thetadatadx-ffi/tests/streaming_config_forwarding.rs b/thetadatadx-ffi/tests/streaming_config_forwarding.rs index d2b9ab1c..fcfe0f83 100644 --- a/thetadatadx-ffi/tests/streaming_config_forwarding.rs +++ b/thetadatadx-ffi/tests/streaming_config_forwarding.rs @@ -29,8 +29,6 @@ use thetadatadx_ffi::{ thetadatadx_config_get_reconnect_wait_rate_limited_ms, thetadatadx_config_get_reconnect_wait_server_restart_ms, thetadatadx_config_get_streaming_connect_timeout_ms, - thetadatadx_config_get_streaming_host_selection, - thetadatadx_config_get_streaming_host_shuffle_seed, thetadatadx_config_get_streaming_io_read_slice_ms, thetadatadx_config_get_streaming_keepalive_idle_secs, thetadatadx_config_get_streaming_keepalive_interval_secs, @@ -48,8 +46,6 @@ use thetadatadx_ffi::{ thetadatadx_config_set_reconnect_wait_rate_limited_ms, thetadatadx_config_set_reconnect_wait_server_restart_ms, thetadatadx_config_set_streaming_connect_timeout_ms, - thetadatadx_config_set_streaming_host_selection, - thetadatadx_config_set_streaming_host_shuffle_seed, thetadatadx_config_set_streaming_io_read_slice_ms, thetadatadx_config_set_streaming_keepalive_idle_secs, thetadatadx_config_set_streaming_keepalive_interval_secs, @@ -71,17 +67,12 @@ fn streaming_transport_fields_round_trip_through_c_abi() { let mut ka_idle = 0u64; let mut ka_interval = 0u64; let mut ka_retries = 0u32; - let mut host_policy = -1i32; - let mut seed_present = false; - let mut seed = 0u64; // SAFETY: `cfg` is the non-null, not-yet-freed handle from // `thetadatadx_config_production`, mutated/read on a single thread; every // out-pointer is a live stack slot for the call. unsafe { - // The scalar streaming knobs are infallible unit-returning - // setters; `host_selection` / `host_shuffle_seed` validate their - // input and return an `i32` status. + // The scalar streaming knobs are infallible unit-returning setters. thetadatadx_config_set_streaming_timeout_ms(cfg, 10_000); thetadatadx_config_set_streaming_connect_timeout_ms(cfg, 5_000); thetadatadx_config_set_streaming_ping_interval_ms(cfg, 1_000); @@ -90,15 +81,6 @@ fn streaming_transport_fields_round_trip_through_c_abi() { thetadatadx_config_set_streaming_keepalive_idle_secs(cfg, 10); thetadatadx_config_set_streaming_keepalive_interval_secs(cfg, 5); thetadatadx_config_set_streaming_keepalive_retries(cfg, 4); - // Host selection: 1 = the non-default preset (round-trips a - // non-zero discriminant so a stuck-at-zero getter is caught). - assert_eq!(thetadatadx_config_set_streaming_host_selection(cfg, 1), 0); - // The host-shuffle seed carries the widened `(has_value, seed)` - // ABI shape so the `Some(_)` presence survives the boundary. - assert_eq!( - thetadatadx_config_set_streaming_host_shuffle_seed(cfg, true, 42), - 0 - ); assert_eq!( thetadatadx_config_get_streaming_timeout_ms(cfg, &mut timeout), @@ -132,14 +114,6 @@ fn streaming_transport_fields_round_trip_through_c_abi() { thetadatadx_config_get_streaming_keepalive_retries(cfg, &mut ka_retries), 0 ); - assert_eq!( - thetadatadx_config_get_streaming_host_selection(cfg, &mut host_policy), - 0 - ); - assert_eq!( - thetadatadx_config_get_streaming_host_shuffle_seed(cfg, &mut seed_present, &mut seed), - 0 - ); } assert_eq!(timeout, 10_000); @@ -150,9 +124,6 @@ fn streaming_transport_fields_round_trip_through_c_abi() { assert_eq!(ka_idle, 10); assert_eq!(ka_interval, 5); assert_eq!(ka_retries, 4); - assert_eq!(host_policy, 1); - assert!(seed_present); - assert_eq!(seed, 42); // SAFETY: `cfg` is owned here and freed exactly once. unsafe { thetadatadx_config_free(cfg) }; diff --git a/thetadatadx-py/python/thetadatadx/__init__.pyi b/thetadatadx-py/python/thetadatadx/__init__.pyi index 85cacfc9..11fecd65 100644 --- a/thetadatadx-py/python/thetadatadx/__init__.pyi +++ b/thetadatadx-py/python/thetadatadx/__init__.pyi @@ -279,10 +279,6 @@ class Config: """Interval, in seconds, between kernel-side TCP keepalive probes on the streaming socket (default 2).""" streaming_keepalive_retries: int """Number of unanswered kernel-side TCP keepalive probes before the streaming socket is declared dead (default 2).""" - streaming_host_selection: Literal["shuffled", "fixed_order"] - """Streaming host-selection order: ``"shuffled"`` (fault-domain-aware per-client shuffle, seedable via :attr:`streaming_host_shuffle_seed`) or ``"fixed_order"``.""" - streaming_host_shuffle_seed: Optional[int] - """Seed for the per-client streaming host shuffle; ``None`` draws a fresh seed each connect.""" @property def market_data_environment(self) -> Literal["PROD", "STAGE"]: """Target market-data environment carried by this configuration: ``"PROD"`` for the production cluster or ``"STAGE"`` for staging. The market-data and streaming channels are selected independently; :meth:`Config.production` / :meth:`Config.stage` (and the ``THETADATA_MARKET_DATA_TYPE`` key on :meth:`Config.from_dotenv`) set the market-data channel, and this is the readback of that selection. Read-only: the selector is chosen by the environment-tier factories, not assigned directly. Mirrors the ``market_data_type`` string the inline :class:`Client` constructor accepts.""" diff --git a/thetadatadx-py/src/_generated/config_accessors.rs b/thetadatadx-py/src/_generated/config_accessors.rs index dbf9220e..3282767a 100644 --- a/thetadatadx-py/src/_generated/config_accessors.rs +++ b/thetadatadx-py/src/_generated/config_accessors.rs @@ -656,28 +656,6 @@ impl Config { guard.reconnect.jitter.as_str() } - /// Set the streaming host-selection policy. Accepts ``"shuffled"`` - /// (default — fault-domain-aware per-client shuffle) or - /// ``"fixed_order"`` (declared order verbatim), case-insensitive. - #[setter] - fn set_streaming_host_selection(&self, policy: &str) -> PyResult<()> { - let parsed = config::HostSelectionPolicy::parse(policy).ok_or_else(|| { - crate::errors::invalid_parameter_err(format!( - "unknown streaming_host_selection: {policy:?} (expected \"shuffled\" or \"fixed_order\")" - )) - })?; - let mut guard = self.inner.lock().unwrap_or_else(|e| e.into_inner()); - guard.streaming.host_selection = parsed; - Ok(()) - } - - /// Current streaming host-selection policy as a lowercase string. - #[getter] - fn get_streaming_host_selection(&self) -> &'static str { - let guard = self.inner.lock().unwrap_or_else(|e| e.into_inner()); - guard.streaming.host_selection.as_str() - } - /// Set the Prometheus exporter port. ``None`` (the default) keeps /// the exporter disabled; an ``int`` binds an HTTP listener whose /// ``/metrics`` endpoint exposes every counter and histogram. @@ -705,22 +683,4 @@ impl Config { guard.metrics.port } - /// Set the streaming host-shuffle seed. ``None`` (default) derives a - /// fresh per-client seed so a fleet shuffles independently; an - /// explicit value makes the shuffled order deterministic — useful - /// for fleet sharding and tests. Ignored under ``"fixed_order"``. - #[setter] - fn set_streaming_host_shuffle_seed(&self, seed: Option) { - let mut guard = self.inner.lock().unwrap_or_else(|e| e.into_inner()); - guard.streaming.host_shuffle_seed = seed; - } - - /// Current ``streaming.host_shuffle_seed`` value (``None`` = per-client - /// entropy). - #[getter] - fn get_streaming_host_shuffle_seed(&self) -> Option { - let guard = self.inner.lock().unwrap_or_else(|e| e.into_inner()); - guard.streaming.host_shuffle_seed - } - } diff --git a/thetadatadx-py/src/fpss_client.rs b/thetadatadx-py/src/fpss_client.rs index e88f2b27..79c11070 100644 --- a/thetadatadx-py/src/fpss_client.rs +++ b/thetadatadx-py/src/fpss_client.rs @@ -103,8 +103,6 @@ impl FpssParams { .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) } } @@ -1126,7 +1124,7 @@ impl StreamingClient { mod tests { use super::*; use thetadatadx::config::{ - HostSelectionPolicy, JitterMode, ReconnectPolicy, + JitterMode, ReconnectPolicy, }; /// Anti-drift guard for the standalone connect path. @@ -1146,8 +1144,6 @@ mod tests { // Streaming: flip every knob away from its production default. config.set_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; @@ -1172,8 +1168,6 @@ mod tests { 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); diff --git a/thetadatadx-py/src/lib.rs b/thetadatadx-py/src/lib.rs index cf030078..26f22a3d 100644 --- a/thetadatadx-py/src/lib.rs +++ b/thetadatadx-py/src/lib.rs @@ -428,9 +428,8 @@ impl Config { // config_surface.toml. // ``DirectConfig.metrics.port`` (``Optional[int]``, exporter port) - // and the ``reconnect.jitter`` / ``streaming.host_selection`` enums - // are the generated ``enum`` / ``option`` accessors in - // config_surface.toml. + // and the ``reconnect.jitter`` enum are the generated ``enum`` / + // ``option`` accessors in config_surface.toml. /// Target market-data environment carried by this configuration: /// ``"PROD"`` for the production cluster or ``"STAGE"`` for staging. diff --git a/thetadatadx-py/tests/test_config_resilience.py b/thetadatadx-py/tests/test_config_resilience.py index c22d0faa..ed933b3f 100644 --- a/thetadatadx-py/tests/test_config_resilience.py +++ b/thetadatadx-py/tests/test_config_resilience.py @@ -123,28 +123,6 @@ def test_streaming_transport_defaults_and_round_trip(): assert cfg.streaming_keepalive_retries == 4 -def test_streaming_host_selection_round_trips_and_rejects_unknown(): - mod = _import_module() - cfg = mod.Config.production() - assert cfg.streaming_host_selection == "shuffled" - cfg.streaming_host_selection = "fixed_order" - assert cfg.streaming_host_selection == "fixed_order" - cfg.streaming_host_selection = "SHUFFLED" - assert cfg.streaming_host_selection == "shuffled" - with pytest.raises(ValueError, match=r"streaming_host_selection"): - cfg.streaming_host_selection = "round_robin" - - -def test_streaming_host_shuffle_seed_round_trips_none_sentinel(): - mod = _import_module() - cfg = mod.Config.production() - assert cfg.streaming_host_shuffle_seed is None - cfg.streaming_host_shuffle_seed = 42 - assert cfg.streaming_host_shuffle_seed == 42 - cfg.streaming_host_shuffle_seed = None - assert cfg.streaming_host_shuffle_seed is None - - # ─── Market-data retry envelope + flatfile jitter ──────────────────── diff --git a/thetadatadx-py/tests/test_no_gil.py b/thetadatadx-py/tests/test_no_gil.py index dfb312e0..fcd3030e 100644 --- a/thetadatadx-py/tests/test_no_gil.py +++ b/thetadatadx-py/tests/test_no_gil.py @@ -212,10 +212,9 @@ def test_fpss_connect_releases_the_gil(monkeypatch) -> None: creds = td.Credentials("user@example.com", "pw") config = td.Config.production() # The env override only rewrites the primary host slot, leaving the - # other production hosts in place; pin fixed-order selection so the - # blackhole primary is the first connect attempt and the call blocks - # there rather than failing fast against a reachable host. - config.streaming_host_selection = "fixed_order" + # other production hosts in place; the client dials hosts in declared + # order, so the blackhole primary is the first connect attempt and the + # call blocks there rather than failing fast against a reachable host. # Bound the connect window so the test stays fast while still being # long enough for the peer thread to accumulate a decisive count. config.streaming_connect_timeout_ms = 1500 diff --git a/thetadatadx-rs/config_surface.toml b/thetadatadx-rs/config_surface.toml index 3d105412..00f26da8 100644 --- a/thetadatadx-rs/config_surface.toml +++ b/thetadatadx-rs/config_surface.toml @@ -1977,8 +1977,8 @@ Current market-data gRPC host. # variant { Variant => N }`); C++ passes the int through; Python / TS # speak the lowercase string via the core enum `parse` / `as_str`. A # rejected int carries `THETADATADX_ERR_INVALID_PARAMETER`. The -# null-handle code is per-setter (`null_err`): jitter / host_selection -# fall back to the untyped `THETADATADX_ERR_OTHER`. +# null-handle code is per-setter (`null_err`): jitter falls back to the +# untyped `THETADATADX_ERR_OTHER`. [[accessor]] symbol = "thetadatadx_config_set_reconnect_jitter" @@ -2081,92 +2081,6 @@ int = 3 rust = "None" label = "none" -[[accessor]] -symbol = "thetadatadx_config_set_streaming_host_selection" -kind = "enum" -param = "policy" -abi_type = "i32" -path = "streaming.host_selection" -enum_type = "thetadatadx::HostSelectionPolicy" -enum_core = "config::HostSelectionPolicy" -null_err = "other" -enum_expected = "expected 0 (Shuffled) or 1 (FixedOrder)" -py_err = 'unknown streaming_host_selection: {policy:?} (expected "shuffled" or "fixed_order")' -ts_err = 'setStreamingHostSelection: unknown policy {policy:?}; expected "shuffled" or "fixed_order"' -doc = """ -Set the streaming host-selection policy. - -- `policy = 0`: Shuffled (default) — fault-domain-aware per-client - shuffle; a fleet spreads across hosts and consecutive failover - attempts cross physical machines. -- `policy = 1`: FixedOrder — use the declared host order verbatim. - -Returns `0` on success. Returns `-1` and sets `thetadatadx_last_error` -when `policy` is outside the documented `{0, 1}` set or `config` -is null. A rejected `policy` value carries -`thetadatadx_last_error_code = THETADATADX_ERR_INVALID_PARAMETER` so an out-of-domain -enum int surfaces the same typed class across every binding. -""" -cpp_doc = """ -Set the streaming host-selection policy: 0=Shuffled (default), -1=FixedOrder. Throws @c thetadatadx::InvalidParameterError on an -out-of-domain policy (and @c thetadatadx::ThetaDataError on a null -handle), routing through the typed leaf the FFI error code -selects. -""" -py_doc = """ -Set the streaming host-selection policy. Accepts ``"shuffled"`` -(default — fault-domain-aware per-client shuffle) or -``"fixed_order"`` (declared order verbatim), case-insensitive. -""" -ts_doc = """ -Set the streaming host-selection policy. Accepts `"shuffled"` -(default — fault-domain-aware per-client shuffle) or -`"fixed_order"` (declared order verbatim), case-insensitive. -""" -py_param = "policy" -ts_param = "policy" -[[accessor.variant]] -int = 0 -rust = "Shuffled" -label = "shuffled" -[[accessor.variant]] -int = 1 -rust = "FixedOrder" -label = "fixed_order" - -[[accessor]] -symbol = "thetadatadx_config_get_streaming_host_selection" -kind = "enum" -param = "out_policy" -abi_type = "i32" -path = "streaming.host_selection" -enum_type = "thetadatadx::HostSelectionPolicy" -enum_core = "config::HostSelectionPolicy" -doc = """ -Read the configured streaming host-selection policy. Same encoding as -`thetadatadx_config_set_streaming_host_selection`. Returns `0` on success, `-1` -if either pointer is null. -""" -cpp_doc = """ -Current streaming host-selection policy (same encoding as the -setter). -""" -py_doc = """ -Current streaming host-selection policy as a lowercase string. -""" -ts_doc = """ -Current streaming host-selection policy as a lowercase string. -""" -[[accessor.variant]] -int = 0 -rust = "Shuffled" -label = "shuffled" -[[accessor.variant]] -int = 1 -rust = "FixedOrder" -label = "fixed_order" - # ── Option config fields (option) ─────────────────────────────────── # # Plain `(has_value, N)` <-> `Option` accessors. The FFI carries the @@ -2256,73 +2170,3 @@ ts_doc = """ Current `metrics.port` setting. `null` means the exporter is disabled; a `number` is the bound port. """ - -[[accessor]] -symbol = "thetadatadx_config_set_streaming_host_shuffle_seed" -kind = "option" -param = "seed" -abi_type = "u64" -path = "streaming.host_shuffle_seed" -doc = """ -Set the streaming host-shuffle seed using the `(has_value, seed)` -widened ABI shape that preserves the `None` sentinel across the C -boundary. - -* `has_value = false` → `None` (default): every client derives a - fresh per-instance seed, so a fleet shuffles independently. - `seed` is ignored. -* `has_value = true` → `Some(seed)`: the shuffled order becomes - deterministic — useful for fleet sharding and tests. - -Ignored under the `FixedOrder` host-selection policy. Returns `0` -on success, `-1` if `config` is null. -""" -cpp_doc = """ -Set the streaming host-shuffle seed using the (has_value, seed) -shape. has_value=false derives a fresh per-client seed; -has_value=true makes the shuffled order deterministic. -""" -py_doc = """ -Set the streaming host-shuffle seed. ``None`` (default) derives a -fresh per-client seed so a fleet shuffles independently; an -explicit value makes the shuffled order deterministic — useful -for fleet sharding and tests. Ignored under ``"fixed_order"``. -""" -ts_doc = """ -Set the streaming host-shuffle seed. `null` (default) derives a -fresh per-client seed so a fleet shuffles independently; an -explicit `bigint` makes the shuffled order deterministic — -useful for fleet sharding and tests. Ignored under -`"fixed_order"`. -""" -py_param = "seed" -ts_param = "seed" - -[[accessor]] -symbol = "thetadatadx_config_get_streaming_host_shuffle_seed" -kind = "option" -param = "out_seed" -abi_type = "u64" -path = "streaming.host_shuffle_seed" -doc = """ -Read the current streaming host-shuffle seed. Same `(has_value, seed)` -ABI as `thetadatadx_config_set_streaming_host_shuffle_seed`: - -* `*out_has_value = false` → `None` (per-client entropy). `*out_seed` is left `0`. -* `*out_has_value = true` → `Some(*out_seed)`. - -Returns `0` on success, `-1` if any pointer is null. -""" -cpp_doc = """ -Read the streaming host-shuffle seed back. Returns @c std::nullopt for -the per-client-entropy sentinel (no pinned seed); returns the -wrapped seed when the shuffled order is deterministic. -""" -py_doc = """ -Current ``streaming.host_shuffle_seed`` value (``None`` = per-client -entropy). -""" -ts_doc = """ -Current `streaming.host_shuffle_seed` value (`null` = per-client -entropy). -""" diff --git a/thetadatadx-rs/src/client.rs b/thetadatadx-rs/src/client.rs index 681621bb..6158ec80 100644 --- a/thetadatadx-rs/src/client.rs +++ b/thetadatadx-rs/src/client.rs @@ -886,8 +886,6 @@ impl Client { .keepalive_idle_secs(config.streaming.keepalive_idle_secs) .keepalive_interval_secs(config.streaming.keepalive_interval_secs) .keepalive_retries(config.streaming.keepalive_retries) - .host_selection(config.streaming.host_selection) - .host_shuffle_seed(config.streaming.host_shuffle_seed) .build() .map_err(crate::error::Error::from)?; let client_arc = Arc::new(client); diff --git a/thetadatadx-rs/src/config/fpss.rs b/thetadatadx-rs/src/config/fpss.rs index 002d4991..12f0c528 100644 --- a/thetadatadx-rs/src/config/fpss.rs +++ b/thetadatadx-rs/src/config/fpss.rs @@ -1,55 +1,5 @@ //! Streaming (TCP) sub-configuration. -/// How the streaming client orders the configured streaming hosts for the -/// initial connect and every reconnect. -/// -/// The production host list spans two physical machines with two ports -/// each. Ordering decides both steady-state load placement (which host -/// a freshly-started client lands on) and failover behaviour (which -/// host a reconnecting client tries next). -#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] -#[non_exhaustive] -pub enum HostSelectionPolicy { - /// Group hosts by hostname (fault domain), shuffle the group order - /// and the ports within each group per client, then interleave - /// across groups. Default. - /// - /// Two effects: a fleet of clients distributes uniformly across - /// the fault domains instead of all dialling the first declared - /// host, and consecutive failover attempts cross fault domains — - /// the second attempt lands on a different physical machine, not a - /// second port on the machine that just failed. - #[default] - Shuffled, - /// Use the declared order verbatim. Escape hatch for deployments - /// that pin traffic to a specific host for locality or compliance - /// reasons. - FixedOrder, -} - -impl HostSelectionPolicy { - /// Canonical lowercase string for this policy, matching the - /// cross-binding encoding. - #[must_use] - pub fn as_str(self) -> &'static str { - match self { - Self::Shuffled => "shuffled", - Self::FixedOrder => "fixed_order", - } - } - - /// Parse the cross-binding string encoding (case-insensitive). - /// Returns `None` for unrecognised input. - #[must_use] - pub fn parse(s: &str) -> Option { - match s.to_ascii_lowercase().as_str() { - "shuffled" => Some(Self::Shuffled), - "fixed_order" => Some(Self::FixedOrder), - _ => None, - } - } -} - /// Streaming client tuning. /// /// The timing knobs (`timeout_ms`, `ping_interval_ms`, @@ -64,8 +14,9 @@ impl HostSelectionPolicy { pub struct StreamingConfig { /// Streaming hosts. /// - /// The connection layer iterates through these on connection - /// failure, in the order produced by [`Self::host_selection`]. + /// The connection layer iterates through these in declared order on + /// connection failure (a reconnect tries the last-known-good host + /// first, then the rest in order). /// Default: ThetaData's NJ `FPSS_NJ_HOSTS`. /// /// Set through [`DirectConfig::set_streaming_hosts`] so the write is @@ -79,21 +30,6 @@ pub struct StreamingConfig { /// [`DirectConfig::streaming_hosts`]: crate::config::DirectConfig::streaming_hosts pub(crate) hosts: Vec<(String, u16)>, - /// Per-client host ordering policy. Default - /// [`HostSelectionPolicy::Shuffled`] — see the enum docs for the - /// fleet-distribution and fault-domain-failover rationale. - pub host_selection: HostSelectionPolicy, - - /// Optional seed for the [`HostSelectionPolicy::Shuffled`] order. - /// - /// `None` (default) derives a fresh per-client seed, so every - /// client instance shuffles independently. Supplying a value makes - /// the order deterministic — useful for fleet sharding (give each - /// deployment slot a stable seed) and for tests that assert a - /// specific order. Ignored under - /// [`HostSelectionPolicy::FixedOrder`]. - pub host_shuffle_seed: Option, - /// Streaming read timeout in milliseconds. /// /// Drives the per-connection initial socket read timeout, the framing @@ -215,8 +151,6 @@ impl StreamingConfig { ("nj-b.thetadata.us".to_string(), 20000), ("nj-b.thetadata.us".to_string(), 20001), ], - host_selection: HostSelectionPolicy::Shuffled, - host_shuffle_seed: None, timeout_ms: 10_000, ring_size: 131_072, ping_interval_ms: 250, @@ -268,8 +202,6 @@ mod tests { assert_eq!(cfg.keepalive_idle_secs, 5); assert_eq!(cfg.keepalive_interval_secs, 2); assert_eq!(cfg.keepalive_retries, 2); - assert_eq!(cfg.host_selection, HostSelectionPolicy::Shuffled); - assert_eq!(cfg.host_shuffle_seed, None); assert_eq!(cfg.consumer_cpu, None); // Kernel-side half-open detection at the defaults: // idle + interval * retries = 5 + 2*2 = 9 seconds. @@ -277,23 +209,4 @@ mod tests { + cfg.keepalive_interval_secs * u64::from(cfg.keepalive_retries); assert_eq!(detection, 9); } - - #[test] - fn host_selection_policy_string_round_trip() { - for policy in [ - HostSelectionPolicy::Shuffled, - HostSelectionPolicy::FixedOrder, - ] { - assert_eq!(HostSelectionPolicy::parse(policy.as_str()), Some(policy)); - } - assert_eq!( - HostSelectionPolicy::parse("SHUFFLED"), - Some(HostSelectionPolicy::Shuffled) - ); - assert_eq!(HostSelectionPolicy::parse("bogus"), None); - assert_eq!( - HostSelectionPolicy::default(), - HostSelectionPolicy::Shuffled - ); - } } diff --git a/thetadatadx-rs/src/config/mod.rs b/thetadatadx-rs/src/config/mod.rs index a80aed75..a04b8a67 100644 --- a/thetadatadx-rs/src/config/mod.rs +++ b/thetadatadx-rs/src/config/mod.rs @@ -56,7 +56,7 @@ pub use env::{ }; pub use environment::{MarketDataEnvironment, StreamingEnvironment}; pub use flatfiles::{bounds as flatfiles_bounds, FlatFilesConfig}; -pub use fpss::{bounds as streaming_bounds, HostSelectionPolicy, StreamingConfig}; +pub use fpss::{bounds as streaming_bounds, StreamingConfig}; pub use mdds::MarketDataConfig; pub(crate) use mdds::DEFAULT_REQUEST_TIMEOUT_SECS; pub use metrics::MetricsConfig; diff --git a/thetadatadx-rs/src/fpss/connection.rs b/thetadatadx-rs/src/fpss/connection.rs index f75fdd2e..e0604e1a 100644 --- a/thetadatadx-rs/src/fpss/connection.rs +++ b/thetadatadx-rs/src/fpss/connection.rs @@ -10,9 +10,9 @@ //! by the transport long before the platform default of 2+ hours //! - Connect timeout: 2 seconds (configurable) //! - Read timeout: configurable (default 10 seconds) -//! - Host order: fault-domain-aware per-client shuffle by default (see -//! [`order_hosts`]), `FixedOrder` escape hatch preserves declaration -//! order +//! - Host order: the declared order (see [`order_hosts`]); the terminal +//! cycles its host list left to right, and a reconnect promotes the +//! last-known-good host to the front //! //! # Implementation //! @@ -23,14 +23,11 @@ use std::net::{TcpStream, ToSocketAddrs}; use std::sync::{Arc, Once}; use std::time::Duration; -use rand::seq::SliceRandom; -use rand::SeedableRng; use rustls::pki_types::ServerName; use rustls::{ClientConfig, ClientConnection, StreamOwned}; use crate::auth::Credentials; use crate::backoff::JitterMode; -use crate::config::HostSelectionPolicy; use crate::config::ReconnectPolicy; use super::pinning::PinnedVerifier; @@ -65,11 +62,9 @@ pub(crate) struct ConnectWithStreamArgs<'a> { pub stream: FpssStream, pub server_addr: String, /// Declared FPSS host list. The initial connect applies - /// [`order_hosts`] with `preferred = None`; reconnects may promote - /// the last stable host while re-running the policy on the tail. + /// [`order_hosts`] with `preferred = None` (declared order); a + /// reconnect promotes the last stable host to the front. pub hosts: &'a [(String, u16)], - pub host_selection: HostSelectionPolicy, - pub host_shuffle_seed: u64, pub ring_size: usize, /// Fixed low-latency event-ring consumer wait strategy /// ([`super::ring::AdaptiveWaitStrategy`]). @@ -122,85 +117,25 @@ fn ensure_rustls_crypto_provider() { }); } -/// Apply the configured [`HostSelectionPolicy`] to the declared host -/// list, producing the per-client connect/failover order. +/// Produce the per-client connect / failover order for the declared +/// host list. /// -/// Under [`HostSelectionPolicy::FixedOrder`] the declared order is -/// preserved verbatim. Under [`HostSelectionPolicy::Shuffled`] (the -/// default) the hosts are grouped by hostname — each hostname is one -/// fault domain — the group order and the ports within each group are -/// shuffled with the supplied seed, and the result interleaves across -/// groups round-robin. Two properties follow: -/// -/// * **Fleet spread** — clients with independent seeds distribute -/// their first connect uniformly across the fault domains instead of -/// all dialling the first declared host. -/// * **Cross-domain failover** — consecutive attempts alternate fault -/// domains, so the second attempt lands on a different physical -/// machine rather than a second port on the machine that just -/// failed. -/// -/// The seed makes the order deterministic: tests and fleet-sharding -/// deployments pass a fixed seed, production defaults derive a fresh -/// per-client seed from process-local entropy. -pub(crate) fn order_hosts( - hosts: &[(String, u16)], - policy: HostSelectionPolicy, - seed: u64, - preferred: Option, -) -> Vec<(String, u16)> { - let preferred = preferred.and_then(|idx| hosts.get(idx).map(|host| (idx, host.clone()))); - let preferred_idx = preferred.as_ref().map(|(idx, _)| *idx); - let mut ordered = match policy { - HostSelectionPolicy::FixedOrder => hosts - .iter() - .enumerate() - .filter(|(idx, _)| preferred_idx != Some(*idx)) - .map(|(_, host)| host.clone()) - .collect(), - // `HostSelectionPolicy` is non_exhaustive; route any future - // variant added without an arm here to the safe default. - _ => { - // Group by hostname, preserving first-seen group order as - // the pre-shuffle baseline. - let mut groups: Vec<(String, Vec)> = Vec::new(); - for (idx, (host, port)) in hosts.iter().enumerate() { - if preferred_idx == Some(idx) { - continue; - } - match groups.iter_mut().find(|(h, _)| h == host) { - Some((_, ports)) => ports.push(*port), - None => groups.push((host.clone(), vec![*port])), - } - } - let mut rng = rand::rngs::StdRng::seed_from_u64(seed); - groups.shuffle(&mut rng); - for (_, ports) in &mut groups { - ports.shuffle(&mut rng); - } - // Round-robin interleave across groups: one port per - // group per round, so consecutive entries cross fault - // domains whenever more than one domain exists. - let mut ordered = Vec::with_capacity(hosts.len()); - let mut round = 0; - loop { - let mut emitted = false; - for (host, ports) in &groups { - if let Some(port) = ports.get(round) { - ordered.push((host.clone(), *port)); - emitted = true; - } - } - if !emitted { - break; - } - round += 1; - } - ordered +/// The declared order is preserved verbatim — the terminal cycles its +/// FPSS host list left to right, and the SDK matches that. On a +/// reconnect the last-known-good host (`preferred`) is tried first, so +/// a client that just lost a working connection re-dials it before +/// walking the rest of the list in declared order. `preferred = None` +/// (the initial cold connect) yields the declared order unchanged. +pub(crate) fn order_hosts(hosts: &[(String, u16)], preferred: Option) -> Vec<(String, u16)> { + let mut ordered = Vec::with_capacity(hosts.len()); + if let Some(host) = preferred.and_then(|idx| hosts.get(idx)) { + ordered.push(host.clone()); + } + for (idx, host) in hosts.iter().enumerate() { + if preferred == Some(idx) { + continue; } - }; - if let Some((_, preferred_host)) = preferred { - ordered.insert(0, preferred_host); + ordered.push(host.clone()); } ordered } @@ -550,142 +485,49 @@ mod tests { } #[test] - fn order_hosts_fixed_order_preserves_declaration() { + fn order_hosts_cold_connect_is_declared_order() { + // The initial connect (`preferred = None`) walks the host list + // in declared order, left to right — matching the terminal. let hosts = production_hosts(); - let ordered = order_hosts(&hosts, HostSelectionPolicy::FixedOrder, 42, None); - assert_eq!(ordered, hosts); - } - - /// The shuffled order is deterministic for a given seed — the - /// load-bearing property for fleet sharding and for this test - /// suite itself. - #[test] - fn order_hosts_shuffled_is_deterministic_per_seed() { - let hosts = production_hosts(); - let a = order_hosts(&hosts, HostSelectionPolicy::Shuffled, 7, None); - let b = order_hosts(&hosts, HostSelectionPolicy::Shuffled, 7, None); - assert_eq!(a, b, "same seed must produce the same order"); - // A different seed must be able to produce a different order; - // with 2 groups x 2 ports there are 8 distinct outcomes, so - // scanning a small seed range must find at least one divergence. - let found_divergent = - (0..32_u64).any(|s| order_hosts(&hosts, HostSelectionPolicy::Shuffled, s, None) != a); - assert!(found_divergent, "shuffle must depend on the seed"); - } - - /// Consecutive entries must alternate fault domains (hostnames) - /// whenever more than one domain exists — the property that makes - /// the first failover land on a different physical machine. - #[test] - fn order_hosts_shuffled_interleaves_fault_domains() { - let hosts = production_hosts(); - for seed in 0..64_u64 { - let ordered = order_hosts(&hosts, HostSelectionPolicy::Shuffled, seed, None); - assert_eq!(ordered.len(), 4, "no hosts may be lost"); - // Same multiset of entries. - let mut sorted_in = hosts.clone(); - let mut sorted_out = ordered.clone(); - sorted_in.sort(); - sorted_out.sort(); - assert_eq!(sorted_in, sorted_out, "shuffle must be a permutation"); - // First two entries cross fault domains. - assert_ne!( - ordered[0].0, ordered[1].0, - "seed {seed}: first failover must cross fault domains; got {ordered:?}" - ); - // Full alternation for the 2x2 production shape. - assert_ne!(ordered[2].0, ordered[3].0); - } + assert_eq!(order_hosts(&hosts, None), hosts); } - /// First-host distribution: across seeds, both fault domains must - /// appear in the first slot — the steady-state load-spreading - /// property. - #[test] - fn order_hosts_shuffled_spreads_first_host_across_domains() { - let hosts = production_hosts(); - let mut first_hosts = std::collections::HashSet::new(); - for seed in 0..64_u64 { - let ordered = order_hosts(&hosts, HostSelectionPolicy::Shuffled, seed, None); - first_hosts.insert(ordered[0].0.clone()); - } - assert_eq!( - first_hosts.len(), - 2, - "both fault domains must appear as the first connect target across seeds" - ); - } - - /// Degenerate shapes: a single host, and asymmetric port counts - /// per domain, must survive the interleave without loss. #[test] fn order_hosts_handles_degenerate_shapes() { let single = vec![("nj-a.thetadata.us".to_string(), 20000)]; - assert_eq!( - order_hosts(&single, HostSelectionPolicy::Shuffled, 1, None), - single - ); - - let asymmetric = vec![ - ("a.example".to_string(), 1), - ("a.example".to_string(), 2), - ("a.example".to_string(), 3), - ("b.example".to_string(), 9), - ]; - let ordered = order_hosts(&asymmetric, HostSelectionPolicy::Shuffled, 5, None); - assert_eq!(ordered.len(), 4); - let mut sorted_in = asymmetric.clone(); - let mut sorted_out = ordered.clone(); - sorted_in.sort(); - sorted_out.sort(); - assert_eq!(sorted_in, sorted_out); + assert_eq!(order_hosts(&single, None), single); + assert_eq!(order_hosts(&[], None), Vec::new()); } #[test] fn order_hosts_reconnect_pins_last_known_good_host_first() { let hosts = production_hosts(); - let ordered = order_hosts(&hosts, HostSelectionPolicy::FixedOrder, 42, Some(2)); + let ordered = order_hosts(&hosts, Some(2)); assert_eq!(ordered[0], hosts[2]); } #[test] - fn order_hosts_reconnect_tail_still_follows_policy() { + fn order_hosts_reconnect_tail_follows_declared_order() { + // After the preferred host, the remaining hosts keep their + // declared order. let hosts = production_hosts(); let preferred = 2; - let ordered = order_hosts(&hosts, HostSelectionPolicy::Shuffled, 17, Some(preferred)); - let tail_hosts: Vec<(String, u16)> = hosts + let ordered = order_hosts(&hosts, Some(preferred)); + let tail_expected: Vec<(String, u16)> = hosts .iter() .enumerate() .filter(|(idx, _)| *idx != preferred) .map(|(_, host)| host.clone()) .collect(); - let tail = order_hosts(&tail_hosts, HostSelectionPolicy::Shuffled, 17, None); assert_eq!(ordered[0], hosts[preferred]); - assert_eq!(&ordered[1..], tail); - } - - #[test] - fn order_hosts_cold_connect_remains_pure_policy() { - let hosts = production_hosts(); - let ordered = order_hosts(&hosts, HostSelectionPolicy::FixedOrder, 42, None); - assert_eq!(ordered, hosts); + assert_eq!(&ordered[1..], tail_expected.as_slice()); + // No host is lost or duplicated. + assert_eq!(ordered.len(), hosts.len()); } #[test] - fn order_hosts_shuffled_tail_still_varies_after_pinning() { + fn order_hosts_out_of_range_preferred_is_ignored() { let hosts = production_hosts(); - let preferred = 2; - let tails: std::collections::HashSet> = (0..32_u64) - .map(|seed| { - let ordered = - order_hosts(&hosts, HostSelectionPolicy::Shuffled, seed, Some(preferred)); - assert_eq!(ordered[0], hosts[preferred]); - ordered[1..].to_vec() - }) - .collect(); - assert!( - tails.len() > 1, - "pinning the first reconnect target must still leave a shuffled tail" - ); + assert_eq!(order_hosts(&hosts, Some(99)), hosts); } } diff --git a/thetadatadx-rs/src/fpss/io_loop/mod.rs b/thetadatadx-rs/src/fpss/io_loop/mod.rs index 1291bcc7..61820bdf 100644 --- a/thetadatadx-rs/src/fpss/io_loop/mod.rs +++ b/thetadatadx-rs/src/fpss/io_loop/mod.rs @@ -46,8 +46,7 @@ use crate::tdbe::types::enums::{RemoveReason, StreamMsgType, StreamResponseType} use crate::auth::Credentials; use crate::backoff::{BackoffSchedule, JitterMode}; use crate::config::{ - HostSelectionPolicy, ReconnectAttemptClass, ReconnectAttemptLimits, ReconnectPolicy, - RATE_LIMITED_JITTER_WINDOW, + ReconnectAttemptClass, ReconnectAttemptLimits, ReconnectPolicy, RATE_LIMITED_JITTER_WINDOW, }; use crate::error::Error; @@ -483,8 +482,6 @@ pub(in crate::fpss) struct IoLoopArgs

{ /// selection policy to this list, optionally pinning the last /// stable host first. pub hosts: Vec<(String, u16)>, - pub host_selection: HostSelectionPolicy, - pub host_shuffle_seed: u64, pub active_subs: ActiveSubs, pub active_full_subs: ActiveFullSubs, /// In-flight subscribe registry keyed by `req_id`. A subscribe records @@ -697,8 +694,6 @@ where replay_pace_ms, creds, hosts, - host_selection, - host_shuffle_seed, active_subs, active_full_subs, pending_subs, @@ -1352,12 +1347,7 @@ where // then re-apply the configured policy to the remaining hosts. // Cold connects and unstable sessions stay pure-policy. let new_stream = { - let ordered_hosts = connection::order_hosts( - &hosts, - host_selection, - host_shuffle_seed, - last_known_good_host, - ); + let ordered_hosts = connection::order_hosts(&hosts, last_known_good_host); let ordered: Vec<(&str, u16)> = ordered_hosts .iter() .map(|(host, port)| (host.as_str(), *port)) diff --git a/thetadatadx-rs/src/fpss/mod.rs b/thetadatadx-rs/src/fpss/mod.rs index 5617b8c2..6dbe71d8 100644 --- a/thetadatadx-rs/src/fpss/mod.rs +++ b/thetadatadx-rs/src/fpss/mod.rs @@ -129,7 +129,7 @@ use std::time::Duration; use crate::auth::Credentials; use crate::backoff::JitterMode; -use crate::config::{HostSelectionPolicy, ReconnectPolicy}; +use crate::config::ReconnectPolicy; use crate::error::Error; use crate::tdbe::types::enums::{RemoveReason, SecType, StreamMsgType}; @@ -442,8 +442,6 @@ pub struct StreamingClientBuilder<'a> { keepalive_idle_secs: u64, keepalive_interval_secs: u64, keepalive_retries: u32, - host_selection: HostSelectionPolicy, - host_shuffle_seed: Option, wait_strategy: ring::AdaptiveWaitStrategy, consumer_cpu: Option, } @@ -480,8 +478,6 @@ impl<'a> StreamingClientBuilder<'a> { keepalive_idle_secs: fpss.keepalive_idle_secs, keepalive_interval_secs: fpss.keepalive_interval_secs, keepalive_retries: fpss.keepalive_retries, - host_selection: fpss.host_selection, - host_shuffle_seed: fpss.host_shuffle_seed, wait_strategy: ring::AdaptiveWaitStrategy::low_latency(), consumer_cpu: fpss.consumer_cpu, } @@ -628,23 +624,6 @@ impl<'a> StreamingClientBuilder<'a> { self } - /// Host-ordering policy for connect + failover. See - /// [`HostSelectionPolicy`]. - #[must_use] - pub fn host_selection(mut self, policy: HostSelectionPolicy) -> Self { - self.host_selection = policy; - self - } - - /// Seed for the shuffled host order; `None` derives a fresh - /// per-client seed. See - /// [`crate::config::StreamingConfig::host_shuffle_seed`]. - #[must_use] - pub fn host_shuffle_seed(mut self, seed: Option) -> Self { - self.host_shuffle_seed = seed; - self - } - /// Connect, authenticate, and start the background I/O and ping /// threads. Returns a ready-to-use [`StreamingClient`]. /// @@ -678,8 +657,6 @@ impl<'a> StreamingClientBuilder<'a> { keepalive_idle_secs: self.keepalive_idle_secs, keepalive_interval_secs: self.keepalive_interval_secs, keepalive_retries: self.keepalive_retries, - host_selection: self.host_selection, - host_shuffle_seed: self.host_shuffle_seed, wait_strategy: self.wait_strategy, consumer_cpu: self.consumer_cpu, } @@ -714,8 +691,6 @@ pub(crate) struct FpssConnectArgs<'a> { pub(crate) keepalive_idle_secs: u64, pub(crate) keepalive_interval_secs: u64, pub(crate) keepalive_retries: u32, - pub(crate) host_selection: HostSelectionPolicy, - pub(crate) host_shuffle_seed: Option, /// Fixed low-latency event-ring consumer wait strategy /// ([`ring::AdaptiveWaitStrategy`]). pub(crate) wait_strategy: ring::AdaptiveWaitStrategy, @@ -797,8 +772,6 @@ struct SpawnArgs<'a, P> { server_addr: String, creds: &'a Credentials, hosts: &'a [(String, u16)], - host_selection: HostSelectionPolicy, - host_shuffle_seed: u64, wait_strategy: ring::AdaptiveWaitStrategy, consumer_cpu: Option, policy: ReconnectPolicy, @@ -1040,8 +1013,6 @@ impl StreamingClient { keepalive_idle_secs, keepalive_interval_secs, keepalive_retries, - host_selection, - host_shuffle_seed, wait_strategy, consumer_cpu, } = args; @@ -1107,10 +1078,8 @@ impl StreamingClient { )); } // Apply the host-selection policy once for the cold connect. - // Reconnects reuse the same seed, optionally pinning the last - // stable host ahead of a policy-ordered tail. - let seed = host_shuffle_seed.unwrap_or_else(crate::backoff::entropy_u64); - let ordered_hosts = connection::order_hosts(hosts, host_selection, seed, None); + // Cold connect: walk the declared host order left to right. + let ordered_hosts = connection::order_hosts(hosts, None); let keepalive = connection::TcpKeepaliveSpec { idle: Duration::from_secs(keepalive_idle_secs), interval: Duration::from_secs(keepalive_interval_secs), @@ -1144,8 +1113,6 @@ impl StreamingClient { stream, server_addr, hosts, - host_selection, - host_shuffle_seed: seed, ring_size, wait_strategy, consumer_cpu, @@ -1182,8 +1149,6 @@ impl StreamingClient { mut stream, server_addr, hosts, - host_selection, - host_shuffle_seed, ring_size, wait_strategy, consumer_cpu, @@ -1341,8 +1306,6 @@ impl StreamingClient { server_addr, creds, hosts, - host_selection, - host_shuffle_seed, wait_strategy, consumer_cpu, policy, @@ -1393,8 +1356,6 @@ impl StreamingClient { server_addr, creds, hosts, - host_selection, - host_shuffle_seed, wait_strategy, consumer_cpu, policy, @@ -1472,8 +1433,6 @@ impl StreamingClient { replay_pace_ms, creds: io_creds, hosts: io_hosts, - host_selection, - host_shuffle_seed, active_subs: io_active_subs, active_full_subs: io_active_full_subs, pending_subs: io_pending_subs, @@ -3152,8 +3111,6 @@ mod builder_tests { assert_eq!(args.keepalive_idle_secs, fpss.keepalive_idle_secs); assert_eq!(args.keepalive_interval_secs, fpss.keepalive_interval_secs); assert_eq!(args.keepalive_retries, fpss.keepalive_retries); - assert_eq!(args.host_selection, fpss.host_selection); - assert_eq!(args.host_shuffle_seed, fpss.host_shuffle_seed); assert_eq!(args.wait_ms, reconnect.wait_ms); assert_eq!(args.wait_max_ms, reconnect.wait_max_ms); assert_eq!(args.wait_rate_limited_ms, reconnect.wait_rate_limited_ms); diff --git a/thetadatadx-rs/src/lib.rs b/thetadatadx-rs/src/lib.rs index 42b59b9c..d327d20d 100644 --- a/thetadatadx-rs/src/lib.rs +++ b/thetadatadx-rs/src/lib.rs @@ -275,9 +275,8 @@ pub use backoff::JitterMode; pub use client::{Client, ConnectionStatus, FlatFiles, StreamSurface, SubscriptionInfo}; pub use client_builder::ClientBuilder; pub use config::{ - DirectConfig, FlatFilesConfig, HostSelectionPolicy, MarketDataEnvironment, - ReconnectAttemptClass, ReconnectAttemptLimits, ReconnectPolicy, RetryPolicy, RuntimeConfig, - StreamingEnvironment, + DirectConfig, FlatFilesConfig, MarketDataEnvironment, ReconnectAttemptClass, + ReconnectAttemptLimits, ReconnectPolicy, RetryPolicy, RuntimeConfig, StreamingEnvironment, }; pub use error::{ AuthErrorKind, ConfigErrorKind, DecodeErrorKind, DecompressErrorKind, Error, GrpcStatusKind, diff --git a/thetadatadx-ts/__tests__/config_resilience.test.mjs b/thetadatadx-ts/__tests__/config_resilience.test.mjs index 2b2934f6..f499093c 100644 --- a/thetadatadx-ts/__tests__/config_resilience.test.mjs +++ b/thetadatadx-ts/__tests__/config_resilience.test.mjs @@ -92,25 +92,6 @@ test("streaming ring size round-trips via BigInt and rejects non-power-of-two", assert.equal(cfg.streamingRingSize, 4_294_967_296n, "rejected value leaves config unchanged"); }); -test("streaming host selection round-trips and rejects unknown", () => { - const cfg = Config.production(); - assert.equal(cfg.streamingHostSelection, "shuffled"); - cfg.setStreamingHostSelection("fixed_order"); - assert.equal(cfg.streamingHostSelection, "fixed_order"); - cfg.setStreamingHostSelection("SHUFFLED"); - assert.equal(cfg.streamingHostSelection, "shuffled"); - assert.throws(() => cfg.setStreamingHostSelection("round_robin"), /shuffled/); -}); - -test("streaming host shuffle seed round-trips the null sentinel", () => { - const cfg = Config.production(); - assert.equal(cfg.streamingHostShuffleSeed, null); - cfg.setStreamingHostShuffleSeed(42n); - assert.equal(cfg.streamingHostShuffleSeed, 42n); - cfg.setStreamingHostShuffleSeed(null); - assert.equal(cfg.streamingHostShuffleSeed, null); -}); - test("retry envelope defaults and round-trip", () => { const cfg = Config.production(); assert.equal(cfg.retryMaxAttempts, 20); diff --git a/thetadatadx-ts/index.d.ts b/thetadatadx-ts/index.d.ts index fbe4bc48..14e8b6ea 100644 --- a/thetadatadx-ts/index.d.ts +++ b/thetadatadx-ts/index.d.ts @@ -573,14 +573,6 @@ export declare class Config { setReconnectJitter(mode: string): void /** Current reconnect jitter mode as a lowercase string. */ get reconnectJitter(): string - /** - * Set the streaming host-selection policy. Accepts `"shuffled"` - * (default — fault-domain-aware per-client shuffle) or - * `"fixed_order"` (declared order verbatim), case-insensitive. - */ - setStreamingHostSelection(policy: string): void - /** Current streaming host-selection policy as a lowercase string. */ - get streamingHostSelection(): string /** * Set the Prometheus exporter port. Pass `null` or `undefined` * to leave the exporter disabled (the default); pass a @@ -595,19 +587,6 @@ export declare class Config { * disabled; a `number` is the bound port. */ get metricsPort(): number | null - /** - * Set the streaming host-shuffle seed. `null` (default) derives a - * fresh per-client seed so a fleet shuffles independently; an - * explicit `bigint` makes the shuffled order deterministic — - * useful for fleet sharding and tests. Ignored under - * `"fixed_order"`. - */ - setStreamingHostShuffleSeed(seed?: bigint | undefined | null): void - /** - * Current `streaming.host_shuffle_seed` value (`null` = per-client - * entropy). - */ - get streamingHostShuffleSeed(): bigint | null } /** diff --git a/thetadatadx-ts/src/_generated/config_accessors.rs b/thetadatadx-ts/src/_generated/config_accessors.rs index c13bdd6d..ab989c1b 100644 --- a/thetadatadx-ts/src/_generated/config_accessors.rs +++ b/thetadatadx-ts/src/_generated/config_accessors.rs @@ -947,34 +947,6 @@ impl Config { Ok(guard.reconnect.jitter.as_str()) } - /// Set the streaming host-selection policy. Accepts `"shuffled"` - /// (default — fault-domain-aware per-client shuffle) or - /// `"fixed_order"` (declared order verbatim), case-insensitive. - #[napi(js_name = "setStreamingHostSelection")] - pub fn set_streaming_host_selection(&self, policy: String) -> napi::Result<()> { - let parsed = config::HostSelectionPolicy::parse(&policy).ok_or_else(|| { - crate::invalid_parameter_err(format!( - "setStreamingHostSelection: unknown policy {policy:?}; expected \"shuffled\" or \"fixed_order\"" - )) - })?; - let mut guard = self - .inner - .lock() - .map_err(|_| napi::Error::from_reason("Config mutex poisoned"))?; - guard.streaming.host_selection = parsed; - Ok(()) - } - - /// Current streaming host-selection policy as a lowercase string. - #[napi(getter, js_name = "streamingHostSelection")] - pub fn streaming_host_selection(&self) -> napi::Result<&'static str> { - let guard = self - .inner - .lock() - .map_err(|_| napi::Error::from_reason("Config mutex poisoned"))?; - Ok(guard.streaming.host_selection.as_str()) - } - /// Set the Prometheus exporter port. Pass `null` or `undefined` /// to leave the exporter disabled (the default); pass a /// `number` to bind an HTTP listener on `0.0.0.0:` when the @@ -1010,39 +982,4 @@ impl Config { Ok(guard.metrics.port.map(u32::from)) } - /// Set the streaming host-shuffle seed. `null` (default) derives a - /// fresh per-client seed so a fleet shuffles independently; an - /// explicit `bigint` makes the shuffled order deterministic — - /// useful for fleet sharding and tests. Ignored under - /// `"fixed_order"`. - #[napi(js_name = "setStreamingHostShuffleSeed")] - pub fn set_streaming_host_shuffle_seed( - &self, - seed: Option, - ) -> napi::Result<()> { - let resolved = match seed { - Some(v) => Some(bigint_to_u64("setStreamingHostShuffleSeed", &v)?), - None => None, - }; - let mut guard = self - .inner - .lock() - .map_err(|_| napi::Error::from_reason("Config mutex poisoned"))?; - guard.streaming.host_shuffle_seed = resolved; - Ok(()) - } - - /// Current `streaming.host_shuffle_seed` value (`null` = per-client - /// entropy). - #[napi(getter, js_name = "streamingHostShuffleSeed")] - pub fn streaming_host_shuffle_seed( - &self, - ) -> napi::Result> { - let guard = self - .inner - .lock() - .map_err(|_| napi::Error::from_reason("Config mutex poisoned"))?; - Ok(guard.streaming.host_shuffle_seed.map(napi::bindgen_prelude::BigInt::from)) - } - } diff --git a/thetadatadx-ts/src/config_class.rs b/thetadatadx-ts/src/config_class.rs index 98cc5579..30962b65 100644 --- a/thetadatadx-ts/src/config_class.rs +++ b/thetadatadx-ts/src/config_class.rs @@ -312,8 +312,8 @@ impl Config { // config_surface.toml (the `ms` / `string` carve-out kinds). // `metrics.port` (`Option` exporter port) and the - // `reconnectJitter` / `streamingHostSelection` enums are the - // generated `enum` / `option` accessors from config_surface.toml. + // `reconnectJitter` enum are the generated `enum` / `option` + // accessors from config_surface.toml. /// Target market-data environment carried by this configuration: /// `"PROD"` for the production cluster or `"STAGE"` for staging. The diff --git a/thetadatadx-ts/src/fpss_client.rs b/thetadatadx-ts/src/fpss_client.rs index a41dc6d3..1a67e3bc 100644 --- a/thetadatadx-ts/src/fpss_client.rs +++ b/thetadatadx-ts/src/fpss_client.rs @@ -354,8 +354,6 @@ impl FpssParams { .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) } } @@ -1329,7 +1327,7 @@ impl StreamingClient { mod tests { use super::*; use thetadatadx::config::{ - HostSelectionPolicy, JitterMode, ReconnectPolicy, + JitterMode, ReconnectPolicy, }; /// Anti-drift guard for the standalone connect path. @@ -1349,8 +1347,6 @@ mod tests { // Streaming: flip every knob away from its production default. config.set_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; @@ -1375,8 +1371,6 @@ mod tests { 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);