chore(fpss): remove the dead wait-strategy escape hatch + hand-rolled splitmix (#1200)

for_each_with_wait_strategy<W> and its streaming::wait re-export module had no caller but their own test — a public generic that can't cross the FFI and that the bindings never use (they drain through for_each_scoped). Remove the method, the re-export, and the test.

seeded_cursor drew its offset from a hand-rolled entropy_u64/splitmix64; use the already-imported rand instead and delete both functions. No behavior change.

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

Author: userFRM

thetadatadx-rs/src/backoff.rs

@@ -133,36 +133,6 @@ pub(crate) fn uniform_duration(lo: Duration, hi: Duration) -> Duration {
     Duration::from_nanos(sampled)
 }
 
-/// Process-local entropy word for seeding shuffles and cursor offsets
-/// where no caller-supplied seed exists. Folds the wall clock with a
-/// process-local counter so two clients constructed in the same
-/// nanosecond still diverge.
-pub(crate) fn entropy_u64() -> u64 {
-    use std::sync::atomic::{AtomicU64, Ordering};
-    static COUNTER: AtomicU64 = AtomicU64::new(0);
-    let tick = COUNTER.fetch_add(1, Ordering::Relaxed);
-    let now = u64::try_from(
-        std::time::SystemTime::now()
-            .duration_since(std::time::UNIX_EPOCH)
-            .map_or(0, |d| d.as_nanos()),
-    )
-    .unwrap_or(u64::MAX);
-    let pid = u64::from(std::process::id());
-    splitmix64(now ^ pid.rotate_left(32) ^ tick.wrapping_mul(0x9E37_79B9_7F4A_7C15))
-}
-
-/// splitmix64 finaliser — documented mixer, used here only to spread
-/// seed bits, never for cryptographic purposes.
-pub(crate) fn splitmix64(mut seed: u64) -> u64 {
-    seed = seed.wrapping_add(0x9E37_79B9_7F4A_7C15);
-    seed ^= seed >> 30;
-    seed = seed.wrapping_mul(0xBF58_476D_1CE4_E5B9);
-    seed ^= seed >> 27;
-    seed = seed.wrapping_mul(0x94D0_49BB_1331_11EB);
-    seed ^= seed >> 31;
-    seed
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -224,11 +194,4 @@ mod tests {
         assert_eq!(JitterMode::parse("bogus"), None);
         assert_eq!(JitterMode::default(), JitterMode::Full);
     }
-
-    #[test]
-    fn entropy_words_diverge() {
-        let a = entropy_u64();
-        let b = entropy_u64();
-        assert_ne!(a, b, "consecutive entropy words must differ");
-    }
 }

thetadatadx-rs/src/fpss/mod.rs

@@ -2089,55 +2089,6 @@ impl StreamingClient {
         }
     }
 
-    /// Drain events through `on_event`, applying a caller-supplied
-    /// [`crate::streaming::wait::WaitStrategy`] on each momentarily
-    /// empty ring instead of the default low-latency wait.
-    ///
-    /// This is the Rust-native bring-your-own-strategy escape hatch: the
-    /// default wait ([`crate::streaming::wait::BusySpinWithSpinLoopHint`]-style
-    /// spin) suits real-time market data, but a Rust caller with an
-    /// exotic backoff (e.g. an adaptive PID-controlled park, or a
-    /// strategy that coordinates with another subsystem) can supply any
-    /// `W: WaitStrategy` here. Use a strategy from
-    /// [`crate::streaming::wait`] (e.g.
-    /// [`crate::streaming::wait::BusySpin`]) or implement the trait on
-    /// your own type.
-    ///
-    /// `W` is monomorphised into the loop, so the per-poll cost is the
-    /// caller's `wait_for` body with no indirection. Delivery semantics
-    /// match [`Self::for_each`]: `on_event` fires exactly once per event
-    /// and the loop returns on terminal shutdown after the ring drains.
-    ///
-    /// # Why Rust-only
-    ///
-    /// `wait_for` fires on every ring-empty poll on the hot path. Routing
-    /// that per-poll callback across the C ABI, the CPython interpreter
-    /// lock, or the JavaScript event loop would add call-boundary
-    /// overhead to the single tightest loop in the consumer — a latency
-    /// regression, not a tuning knob. The bindings therefore run the
-    /// fixed low-latency wait with no override.
-    pub fn for_each_with_wait_strategy<W, F>(&self, mut on_event: F, strategy: W) -> PollOutcome
-    where
-        W: crate::streaming::wait::WaitStrategy,
-        F: FnMut(&StreamEvent),
-    {
-        // The drain owner + CPU pin are recorded inside `poll_batch`, after
-        // the staging lock is acquired, so the identity reflects the proven
-        // drainer rather than whichever thread merely entered this loop.
-        loop {
-            match self.poll_batch(&mut on_event) {
-                // Return the terminal outcome so the caller can tell a clean
-                // shutdown from an I/O-thread fault (`Failed`).
-                terminal @ (PollOutcome::Shutdown | PollOutcome::Failed) => return terminal,
-                // `Busy` cannot arise from this loop's own poll (it holds no
-                // lock when it polls); back off and retry defensively in case
-                // an `on_event` callback re-entered a drain.
-                PollOutcome::Drained(0) | PollOutcome::Busy => strategy.wait_for(0),
-                PollOutcome::Drained(_) => {}
-            }
-        }
-    }
-
     /// Polymorphic subscribe — wire-level entry point.
     ///
     /// Accepts a typed [`protocol::Subscription`] value built via
@@ -4144,35 +4095,6 @@ mod ring_occupancy_tests {
         );
     }
 
-    /// The bring-your-own-strategy drain is reachable through the
-    /// crate-owned wait-strategy re-export, so a Rust caller never names
-    /// the underlying ring crate. Dropping the producer terminates the
-    /// drain; `BusySpin`'s `wait_for` is a no-op, so the loop spins only
-    /// until the ring reports shutdown.
-    #[test]
-    fn for_each_with_wait_strategy_accepts_crate_owned_preset() {
-        use crate::streaming::wait::BusySpin;
-
-        let (client, mut producer) = StreamingClient::for_ring_occupancy_test(64);
-        for _ in 0..3 {
-            assert!(
-                publish_one(&mut producer),
-                "fresh ring must accept a publish"
-            );
-        }
-        // Drop the producer so the poller observes terminal shutdown once
-        // the three published events drain, and the loop returns.
-        drop(producer);
-
-        let mut delivered = 0usize;
-        client.for_each_with_wait_strategy(|_event| delivered += 1, BusySpin);
-
-        assert_eq!(
-            delivered, 3,
-            "every published event must be delivered before shutdown returns"
-        );
-    }
-
     /// A faulted I/O thread — the `io_loop` unwound and its fault guard set
     /// `io_faulted` before the ring published its shutdown sequence — must
     /// surface through the CALLBACK drain path, not read as a clean shutdown.

thetadatadx-rs/src/grpc/pool.rs

@@ -209,9 +209,8 @@ impl ChannelPool {
 /// together after an outage lands its entire first burst on the same
 /// upstream connection.
 fn seeded_cursor(len: usize) -> AtomicUsize {
-    let seed = crate::backoff::entropy_u64();
-    let offset = (seed as usize) % len.max(1);
-    AtomicUsize::new(offset)
+    use rand::RngExt;
+    AtomicUsize::new(rand::rng().random_range(0..len.max(1)))
 }
 
 /// Pre-dispatch reservation on a pooled [`Channel`].

thetadatadx-rs/src/lib.rs

@@ -344,38 +344,6 @@ pub mod streaming {
     #[cfg(feature = "arrow")]
     #[doc(hidden)]
     pub use crate::fpss::batch_schema::stream_batch_schema;
-
-    /// Consumer wait strategies for the streaming ring.
-    ///
-    /// The streaming client drains the ring with a fixed low-latency
-    /// wait, which every language binding uses. A Rust caller that needs
-    /// an exotic backoff can instead supply any type implementing
-    /// `WaitStrategy` to
-    /// [`crate::streaming::StreamingClient::for_each_with_wait_strategy`]. The
-    /// strategy is monomorphised into the drain loop, so the per-poll
-    /// cost is the caller's `wait_for` body with no indirection.
-    ///
-    /// `BusySpin` is the lowest-latency strategy (a true busy spin);
-    /// `BusySpinWithSpinLoopHint` adds a `spin_loop` hint so the core
-    /// can save power or switch hyper-threads; `Sleep` parks the thread
-    /// for a fixed duration between polls.
-    ///
-    /// ```rust,ignore
-    /// use thetadatadx::streaming::wait::BusySpin;
-    ///
-    /// client.for_each_with_wait_strategy(
-    ///     |event| { /* handle event */ },
-    ///     BusySpin,
-    /// );
-    /// ```
-    pub mod wait {
-        // VOCAB-OK: re-exporting the ring's wait-strategy surface under a
-        // crate-owned path so callers never name the underlying ring
-        // crate in their own `use` statements or trait bounds.
-        pub use disruptor::wait_strategies::{
-            BusySpin, BusySpinWithSpinLoopHint, Sleep, WaitStrategy,
-        };
-    }
 }
 
 // ─── Market-data queries ──────────────────────────────────────────────────────