feat(streaming): expose a configurable wait strategy across every binding (#806)

* feat(streaming): expose a configurable wait strategy across every binding

The streaming event-ring consumer applies a wait strategy on every ring-empty poll; this was previously fixed. A `StreamingWaitStrategy` preset now selects the latency-vs-CPU trade-off across every binding: `LowLatency` (the default, which reproduces the previous fixed behaviour byte-for-byte — spin, yield, `spin_loop` hint, never sleeps), `Balanced` (a short spin then a brief timed park), `Efficient` (a longer park for the lowest idle CPU), and `BusySpin` (pure spin, pinning a core). The spin, yield, and park counts tune independently and are clamped to sane bounds.

The preset enum plus the numeric `wait_spin_iters` / `wait_yield_iters` / `wait_park_us` knobs are the FFI-safe form of the override, exposed identically on Python, TypeScript, C++, and the C ABI (with `THETADATADX_WAIT_*` selectors), and enforced cross-binding by the parity guard. Rust additionally gains a zero-cost generic `StreamingClient::for_each_with_wait_strategy` that accepts any user `WaitStrategy` impl; that override is Rust-only by design, because `wait_for` fires on every ring-empty poll and routing a per-poll callback across the C ABI, the CPython lock, or the JavaScript event loop would be a latency regression rather than a tuning knob.

The same surface adds an optional `consumer_cpu` knob that pins the tick-consumer drain thread to a CPU core for deterministic, low-jitter delivery; `None` (a negative sentinel on the C ABI) is the default and leaves the thread under the OS scheduler, so behaviour is unchanged unless a client opts in. An out-of-range or offline core is a best-effort no-op at the affinity layer rather than a hard error. The config-derived strategy threads through the builder into both the ring build site and the consumer-side blocking drain; the disruptor builder erases the wait-strategy type after build, so the poller and producer return types are unchanged and the ring API stays stable.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* test(ffi): name the per-call SAFETY invariant in the wait-strategy test

The wait-strategy C ABI round-trip test repeated a stamped "see the module-level note" SAFETY comment across several unsafe blocks, which the SAFETY-comment hygiene gate rejects. Rewrite each to name the actual invariant — the live, unfreed cfg handle and the valid stack out-params for that specific call.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* fix(streaming): pin the consumer core on the wait-strategy override path and de-flake the preset test

The Rust-only for_each_with_wait_strategy override drained the ring without applying the configured consumer-core affinity, so a caller that set consumer_cpu and used the override lost the pin; call pin_consumer_once at its drain entry like the other drain paths so the config knob applies consistently regardless of which drive API is used.

Replace the fragile absolute upper-bound timing assertion on the never-sleeping low-latency wait_for with a relative comparison against the Balanced preset's real park, so the test does not flake when a loaded host stretches the yield phase.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* docs: remove a third-party project reference from the historical notes

Drop a parenthetical naming an external project's repository layout from the historical changelog, its rendered mirror, and the matching release note. The repository's structure stands on its own description; referencing another project's layout adds nothing and does not belong in public release prose.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* fix: correct the wait-strategy override doc link and sync tool lockfiles

Point the streaming wait-strategy docstring at the real `for_each_with_wait_strategy` method (the rustdoc intra-doc-link gate rejected the stale `start_streaming_with_wait_strategy` name), and resync the tools/mcp and tools/server lockfiles so the excluded-workspace `--locked` builds match their manifests after the new affinity dependency.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: userFRM

.github/release-notes/v9.1.0.md

@@ -74,8 +74,8 @@ the user callback directly from the event-delivery thread.
   TypeScript public-surface drift relative to the runtime SDKs now
   surfaces as a pre-merge agreement failure rather than going
   unnoticed.
-- Repo hygiene pass. Root tree trimmed to standard institutional
-  shape (matches the databento-rs layout): moved `ROADMAP.md` →
+- Repo hygiene pass. Root tree trimmed to a standard institutional
+  layout: moved `ROADMAP.md` →
   `docs/`, moved `config.default.toml` into the `thetadatadx` crate,
   deleted unused `cliff.toml`. Architecture ADRs inlined into
   source-code comments at their relevant locations;

CHANGELOG.md

@@ -39,6 +39,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Configurable streaming latency tuning across every binding: a `StreamingWaitStrategy` event-ring wait-strategy preset (`LowLatency` default, which preserves the previous fixed behaviour, plus `Balanced`, `Efficient`, and `BusySpin`) with independent spin / yield / park tuning, and an optional `consumer_cpu` knob that pins the tick-consumer thread to a CPU core (`None` default leaves it under the OS scheduler); Rust additionally exposes a zero-cost generic `StreamingClient::for_each_with_wait_strategy` override that accepts any user `WaitStrategy` impl, with the preset enum plus numeric tuning as the FFI-safe form on Python (`wait_strategy` / `wait_spin_iters` / `wait_yield_iters` / `wait_park_us` / `consumer_cpu`), TypeScript (`waitStrategy` / `setWaitStrategy` and the matching numeric and `consumerCpu` setters/getters), C++ (`set_wait_strategy` / `get_wait_strategy` and the numeric / `consumer_cpu` forwarders), and the C ABI (`thetadatadx_config_set_wait_strategy` with `THETADATADX_WAIT_*` selectors, the numeric setters/getters, and `thetadatadx_config_set_consumer_cpu` with a negative `THETADATADX_CONSUMER_CPU_UNPINNED` sentinel).
 - Epoch-instant accessors on every row that carries `date` plus a milliseconds-of-day column, computed on read (raw integer fields stay primary): Rust methods and Python properties named `timestamp_ms` for the bare `ms_of_day` column and `<prefix>_timestamp_ms` for prefixed columns (`created_timestamp_ms`, `last_trade_timestamp_ms`, `underlying_timestamp_ms`, `quote_timestamp_ms`), the C function `thetadatadx_timestamp_ms(date, ms_of_day)`, and the C++ `thetadatadx::timestamp_ms` wrapper. All return Unix epoch milliseconds (UTC, DST-aware) and signal absent dates (`None` / `-1`).
 - `thetadatadx::time::date_ms_to_epoch_ms` — the DST-aware inverse of the epoch-to-Eastern split, shared by every accessor above.
 - `thetadatadx::CalendarStatus` — the exported calendar day-type enum with `as_str()` / `from_wire_text()` / `from_code()` / `is_open()`.
@@ -617,8 +618,8 @@ code does not change.
   surfaces as a pre-merge agreement failure rather than going
   unnoticed until a downstream consumer hit the missing / extra
   field.
-- Repo hygiene pass. Root tree trimmed to standard institutional shape
-  (matches the databento-rs layout): moved `ROADMAP.md` → `docs/`,
+- Repo hygiene pass. Root tree trimmed to a standard institutional
+  layout: moved `ROADMAP.md` → `docs/`,
   moved `config.default.toml` into the `thetadatadx` crate, deleted unused
   `cliff.toml`. Architecture ADRs inlined into source-code comments at
   their relevant locations; `docs/architecture/` removed. Generated

Cargo.lock

@@ -188,6 +188,13 @@ regex = { version = "1.12.3", optional = true }
 # Lock-free ring buffer for FPSS event dispatch # VOCAB-OK: crate name in dep
 disruptor = "4.2.0" # VOCAB-OK: actual dep crate name
 
+# CPU core pinning for the streaming event-ring consumer thread
+# (optional `consumer_cpu` tuning knob). Already in the dependency tree
+# via disruptor; named directly so the drain loop can pin the real
+# consumer thread, which the disruptor builder cannot reach in polling
+# mode. # VOCAB-OK: crate name in dep
+core_affinity = "0.8.1" # VOCAB-OK: actual dep crate name
+
 # Optional DataFrame / Arrow extension traits (`frames` module).
 # polars: minimal feature set — just enough to build DataFrame from Series.
 # No lazy, no I/O, no parquet, no SQL, no compute kernels.

crates/thetadatadx/Cargo.toml

@@ -91,7 +91,7 @@ use thetadatadx::fpss::{StreamControl, StreamEvent};
 // from the crate under test, not user-facing prose.
 #[cfg(feature = "__test-helpers")]
 use thetadatadx::fpss::__test_internals::{
-    build_poller_producer, Polling, RingCursors, RingProducer,
+    build_poller_producer, AdaptiveWaitStrategy, Polling, RingCursors, RingProducer,
 };
 
 /// Number of events shipped through the pipeline per criterion sample.
@@ -320,7 +320,11 @@ fn run_disruptor_production_ctor() -> (u64, u64) {
     // Shared occupancy cursors the production adapter records into — the
     // exact pair `StreamingClient::ring_occupancy` samples in the live client.
     let cursors = Arc::new(RingCursors::new());
-    let (mut producer, mut poller) = build_poller_producer(RING_SIZE, Arc::clone(&cursors));
+    let (mut producer, mut poller) = build_poller_producer(
+        RING_SIZE,
+        Arc::clone(&cursors),
+        AdaptiveWaitStrategy::low_latency(),
+    );
 
     // Producer thread: publish via the instrumented adapter. Each
     // successful `try_publish` records the published sequence into the

crates/thetadatadx/benches/streaming_channels.rs

@@ -332,6 +332,13 @@ impl Client {
         let client = StreamingClient::builder(&self.creds, &config.streaming.hosts)
             .ring_size(config.streaming.ring_size)
             .flush_mode(config.streaming.flush_mode)
+            .wait_strategy(config.streaming.wait_strategy)
+            .wait_strategy_tuning(
+                config.streaming.wait_spin_iters,
+                config.streaming.wait_yield_iters,
+                config.streaming.wait_park_us,
+            )
+            .consumer_cpu(config.streaming.consumer_cpu)
             .reconnect_policy(config.reconnect.policy.clone())
             .reconnect_wait_ms(config.reconnect.wait_ms)
             .reconnect_wait_max_ms(config.reconnect.wait_max_ms)

crates/thetadatadx/src/client.rs

@@ -62,6 +62,92 @@ impl HostSelectionPolicy {
     }
 }
 
+/// Wait strategy for the streaming event-ring consumer — the
+/// latency-vs-CPU knob applied on every ring-empty poll.
+///
+/// The consumer drains the lock-free event ring; when it momentarily
+/// finds the ring empty it runs this strategy before re-checking. The
+/// preset picks a point on the latency-vs-CPU curve:
+///
+/// - [`StreamingWaitStrategy::LowLatency`] (default): spin then yield
+///   then a `spin_loop` hint, never sleeps. Lowest latency, highest
+///   idle CPU. Reproduces the historical fixed behaviour.
+/// - [`StreamingWaitStrategy::Balanced`]: short spin then a brief park.
+///   Low idle CPU, ~one park interval of added tail latency.
+/// - [`StreamingWaitStrategy::Efficient`]: minimal spin then a longer
+///   park. Lowest idle CPU.
+/// - [`StreamingWaitStrategy::BusySpin`]: pure spin, no yield or sleep.
+///   Absolute minimum latency; pins a core while idle.
+///
+/// The spin / yield / park counts are tuned independently via
+/// [`StreamingConfig::wait_spin_iters`], [`StreamingConfig::wait_yield_iters`],
+/// and [`StreamingConfig::wait_park_us`].
+#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
+#[non_exhaustive]
+pub enum StreamingWaitStrategy {
+    /// Spin, yield, then a `spin_loop` hint; never sleeps. Lowest
+    /// latency, highest idle CPU. Default.
+    #[default]
+    LowLatency,
+    /// Short spin then a brief timed park. Low idle CPU, slightly higher
+    /// tail latency.
+    Balanced,
+    /// Minimal spin then a longer timed park. Lowest idle CPU.
+    Efficient,
+    /// Pure spin, no yield or sleep. Absolute minimum latency; pins a
+    /// core while the ring is idle.
+    BusySpin,
+}
+
+impl StreamingWaitStrategy {
+    /// Canonical lowercase string for this strategy, matching the
+    /// cross-binding encoding.
+    #[must_use]
+    pub fn as_str(self) -> &'static str {
+        match self {
+            Self::LowLatency => "low_latency",
+            Self::Balanced => "balanced",
+            Self::Efficient => "efficient",
+            Self::BusySpin => "busy_spin",
+        }
+    }
+
+    /// Parse the cross-binding string encoding (case-insensitive).
+    /// Returns `None` for unrecognised input.
+    #[must_use]
+    pub fn parse(s: &str) -> Option<Self> {
+        match s.to_ascii_lowercase().as_str() {
+            "low_latency" => Some(Self::LowLatency),
+            "balanced" => Some(Self::Balanced),
+            "efficient" => Some(Self::Efficient),
+            "busy_spin" => Some(Self::BusySpin),
+            _ => None,
+        }
+    }
+}
+
+impl std::fmt::Display for StreamingWaitStrategy {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(self.as_str())
+    }
+}
+
+impl std::str::FromStr for StreamingWaitStrategy {
+    type Err = crate::error::Error;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        Self::parse(s).ok_or_else(|| {
+            crate::error::Error::config_invalid(
+                "streaming.wait_strategy",
+                format!(
+                    "wait_strategy must be one of \
+                     \"low_latency\", \"balanced\", \"efficient\", \"busy_spin\"; got {s:?}"
+                ),
+            )
+        })
+    }
+}
+
 /// Streaming client tuning.
 ///
 /// The timing knobs (`timeout_ms`, `ping_interval_ms`,
@@ -197,6 +283,46 @@ pub struct StreamingConfig {
     ///   latency, higher syscall overhead.
     pub flush_mode: StreamingFlushMode,
 
+    /// Wait strategy for the event-ring consumer — the latency-vs-CPU
+    /// knob applied on every ring-empty poll.
+    ///
+    /// - [`StreamingWaitStrategy::LowLatency`] (default): never sleeps,
+    ///   lowest latency, highest idle CPU.
+    /// - [`StreamingWaitStrategy::Balanced`]: brief park, low idle CPU.
+    /// - [`StreamingWaitStrategy::Efficient`]: longer park, lowest idle CPU.
+    /// - [`StreamingWaitStrategy::BusySpin`]: pure spin, pins a core.
+    pub wait_strategy: StreamingWaitStrategy,
+
+    /// Spin iterations the wait strategy busy-waits before yielding /
+    /// parking. Higher values trade idle CPU for lower wake latency.
+    /// Default `100`. Clamped to a sane upper bound when applied.
+    pub wait_spin_iters: u32,
+
+    /// `thread::yield_now()` iterations after the spin phase, before any
+    /// park. Higher values smooth brief inter-burst gaps at slightly
+    /// higher idle CPU. Default `10`. Clamped when applied.
+    pub wait_yield_iters: u32,
+
+    /// Park interval (microseconds) for the parking wait strategies
+    /// ([`StreamingWaitStrategy::Balanced`] /
+    /// [`StreamingWaitStrategy::Efficient`]). Larger values lower idle
+    /// CPU at the cost of added tail latency; inert under
+    /// [`StreamingWaitStrategy::LowLatency`] /
+    /// [`StreamingWaitStrategy::BusySpin`], which never sleep. Default
+    /// `50`. Clamped when applied.
+    pub wait_park_us: u64,
+
+    /// Optional CPU core to pin the streaming event-ring consumer thread
+    /// to.
+    ///
+    /// `None` (default) leaves the consumer under the OS scheduler — the
+    /// historical behaviour. `Some(core_id)` pins the tick-consumer
+    /// thread to that core for deterministic, low-jitter delivery; pair
+    /// with an isolated core (e.g. `isolcpus`) for best results. An
+    /// out-of-range or offline core is a best-effort no-op at the
+    /// affinity layer (a `warn` is logged) rather than a hard error.
+    pub consumer_cpu: Option<usize>,
+
     /// Whether to derive OHLCVC bars locally from trade events.
     ///
     /// When `true` (default), the streaming client emits derived `StreamData::Ohlcvc`
@@ -228,9 +354,39 @@ impl StreamingConfig {
             keepalive_interval_secs: 2,
             keepalive_retries: 2,
             flush_mode: StreamingFlushMode::Batched,
+            wait_strategy: StreamingWaitStrategy::LowLatency,
+            wait_spin_iters: 100,
+            wait_yield_iters: 10,
+            wait_park_us: 50,
+            consumer_cpu: None,
             derive_ohlcvc: true,
         }
     }
+
+    /// Build the event-ring consumer wait strategy from the configured
+    /// mode and tuning knobs.
+    ///
+    /// The [`StreamingWaitStrategy::LowLatency`] default with the default
+    /// tuning reproduces the historical fixed FPSS strategy byte-for-byte
+    /// (100 spins / 10 yields / trailing `spin_loop` hint, never sleeps).
+    /// Out-of-range tuning is clamped to the wait strategy's documented
+    /// bounds.
+    #[must_use]
+    pub(crate) fn build_wait_strategy(&self) -> crate::fpss::ring::AdaptiveWaitStrategy {
+        use crate::fpss::ring::{AdaptiveWaitStrategy, WaitMode};
+        let mode = match self.wait_strategy {
+            StreamingWaitStrategy::LowLatency => WaitMode::LowLatency,
+            StreamingWaitStrategy::Balanced => WaitMode::Balanced,
+            StreamingWaitStrategy::Efficient => WaitMode::Efficient,
+            StreamingWaitStrategy::BusySpin => WaitMode::BusySpin,
+        };
+        AdaptiveWaitStrategy::from_mode(
+            mode,
+            self.wait_spin_iters,
+            self.wait_yield_iters,
+            self.wait_park_us,
+        )
+    }
 }
 
 /// Validation bounds for the wired streaming knobs. Out-of-range values
@@ -250,6 +406,19 @@ pub mod bounds {
     pub const KEEPALIVE_INTERVAL_SECS: std::ops::RangeInclusive<u64> = 1..=75;
     /// Allowed range for [`super::StreamingConfig::keepalive_retries`].
     pub const KEEPALIVE_RETRIES: std::ops::RangeInclusive<u32> = 1..=10;
+    /// Allowed range for [`super::StreamingConfig::wait_spin_iters`]. The
+    /// ceiling caps a misconfiguration from turning the spin phase into a
+    /// multi-millisecond busy-wait.
+    pub const WAIT_SPIN_ITERS: std::ops::RangeInclusive<u32> =
+        0..=crate::fpss::ring::AdaptiveWaitStrategy::MAX_SPIN_ITERS;
+    /// Allowed range for [`super::StreamingConfig::wait_yield_iters`].
+    pub const WAIT_YIELD_ITERS: std::ops::RangeInclusive<u32> =
+        0..=crate::fpss::ring::AdaptiveWaitStrategy::MAX_YIELD_ITERS;
+    /// Allowed range for [`super::StreamingConfig::wait_park_us`], in
+    /// microseconds. The ceiling caps a misconfiguration from parking the
+    /// consumer for seconds at a time.
+    pub const WAIT_PARK_US: std::ops::RangeInclusive<u64> =
+        0..=crate::fpss::ring::AdaptiveWaitStrategy::MAX_PARK_US;
 }
 
 impl Default for StreamingConfig {
@@ -274,6 +443,11 @@ mod tests {
         assert_eq!(cfg.keepalive_retries, 2);
         assert_eq!(cfg.host_selection, HostSelectionPolicy::Shuffled);
         assert_eq!(cfg.host_shuffle_seed, None);
+        assert_eq!(cfg.wait_strategy, StreamingWaitStrategy::LowLatency);
+        assert_eq!(cfg.wait_spin_iters, 100);
+        assert_eq!(cfg.wait_yield_iters, 10);
+        assert_eq!(cfg.wait_park_us, 50);
+        assert_eq!(cfg.consumer_cpu, None);
         // Kernel-side half-open detection at the defaults:
         // idle + interval * retries = 5 + 2*2 = 9 seconds.
         let detection = cfg.keepalive_idle_secs
@@ -299,4 +473,58 @@ mod tests {
             HostSelectionPolicy::Shuffled
         );
     }
+
+    #[test]
+    fn wait_strategy_string_round_trip() {
+        use std::str::FromStr;
+        for s in [
+            StreamingWaitStrategy::LowLatency,
+            StreamingWaitStrategy::Balanced,
+            StreamingWaitStrategy::Efficient,
+            StreamingWaitStrategy::BusySpin,
+        ] {
+            assert_eq!(StreamingWaitStrategy::parse(s.as_str()), Some(s));
+            assert_eq!(s.to_string(), s.as_str());
+            assert_eq!(StreamingWaitStrategy::from_str(s.as_str()).unwrap(), s);
+        }
+        assert_eq!(
+            StreamingWaitStrategy::parse("BUSY_SPIN"),
+            Some(StreamingWaitStrategy::BusySpin)
+        );
+        assert_eq!(StreamingWaitStrategy::parse("bogus"), None);
+        assert!(StreamingWaitStrategy::from_str("bogus").is_err());
+        assert_eq!(
+            StreamingWaitStrategy::default(),
+            StreamingWaitStrategy::LowLatency
+        );
+    }
+
+    #[test]
+    fn build_wait_strategy_default_is_low_latency() {
+        let cfg = StreamingConfig::production_defaults();
+        // The default config must reproduce the historical fixed strategy.
+        let built = cfg.build_wait_strategy();
+        let expected = crate::fpss::ring::AdaptiveWaitStrategy::low_latency();
+        // Both are LowLatency with the 100/10 tuning; compare the public
+        // preset constructor against the config-built strategy via their
+        // observable `wait_for` (LowLatency never sleeps).
+        use disruptor::wait_strategies::WaitStrategy;
+        let t0 = std::time::Instant::now();
+        built.wait_for(0);
+        expected.wait_for(0);
+        // LowLatency never parks, so two calls stay well under a
+        // millisecond — guards against the default silently parking.
+        assert!(t0.elapsed() < std::time::Duration::from_millis(5));
+    }
+
+    #[test]
+    fn build_wait_strategy_clamps_out_of_range_tuning() {
+        let mut cfg = StreamingConfig::production_defaults();
+        cfg.wait_strategy = StreamingWaitStrategy::Balanced;
+        cfg.wait_spin_iters = u32::MAX;
+        cfg.wait_yield_iters = u32::MAX;
+        cfg.wait_park_us = u64::MAX;
+        // Must not panic; clamping happens inside `from_mode`.
+        let _ = cfg.build_wait_strategy();
+    }
 }