fix(streaming): harden FPSS subscription tracking, ring defaults, and credential validation (#809)

* fix(streaming): harden FPSS subscription tracking, ring defaults, and credential validation

Oversized streaming credentials now return a typed invalid-parameter configuration error before the connection opens instead of panicking. The credentials payload builder validates the total length against the 255-byte frame limit up front and names both the limit and the actual size; every connect and auth call site propagates the result.

Active subscriptions are de-duplicated by contract and by full-stream security type, so a repeated subscribe no longer accumulates duplicate tracked entries that replay multiple times on reconnect. Unsubscribe keeps its remove-once semantics. The command channel to the I/O worker is now bounded: the heartbeat takes natural backpressure with a blocking send and the public subscribe and unsubscribe methods return a typed queue-full error rather than dropping commands or growing the queue without limit.

The direct streaming builder now defaults its buffer to the production streaming buffer size, so a builder-constructed client gets production-grade headroom for large streams by default; callers can still set a smaller size explicitly. The streaming backpressure documentation is corrected to state that newly arriving events are dropped when the buffer is full, not the oldest buffered events.

Closes #803
Closes #804
Closes #805

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

* fix(bench): consume the now-fallible build_credentials_payload result

build_credentials_payload returns Result since it validates the
credential lengths against the wire frame bounds, so discarding its
value trips unused-must-use under -D warnings on the benches target.

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

---------

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

Author: userFRM

CHANGELOG.md

@@ -58,6 +58,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Changed
 
 - The `tdbe` time-and-calendar crate is folded into `thetadatadx` as a private internal module, so the workspace builds and publishes a single `thetadatadx` artifact. The curated public surface is unchanged (`TradeTick`, `greeks::all_greeks`, `SecType`, `utils::conditions`, and the calendar types are reached at `thetadatadx::*` paths); items that were previously foreign-public — `PriceError`, `MAX_PRICE_TYPE`, `greeks::Error` — are re-exported at public crate paths, and the data-layer internals (DST epoch math, canonical-JSON helpers, the FIT / FIE codecs) stay behind the existing `__internal` feature.
+- The direct `StreamingClientBuilder` now defaults its streaming buffer to the same size as the production streaming configuration, so a builder-constructed client gets production-grade overflow headroom by default for large streams (10k-15k option contracts plus full trade streams) instead of a much smaller buffer that drops events under market bursts; set `.ring_size(..)` to choose a smaller footprint. The streaming backpressure documentation is also corrected to state that newly arriving events are dropped when the buffer is full, not the oldest buffered events.
 
 ### Fixed
 
@@ -70,6 +71,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Python — the standalone `StreamingClient` streaming connect, reconnect, and subscribe / unsubscribe paths release the GIL across their blocking I/O (the TLS connect and handshake, and the per-subscription wire write), so other Python threads keep running while a connect or subscribe is in flight; the typed exception raised on failure is unchanged.
 - The standalone Python and TypeScript streaming clients now forward the full streaming and reconnect config, so every tuning knob — including the wait strategy and consumer-core affinity, host selection, watchdog and keepalive cadences, and the reconnect backoff and replay pacing — is honored, matching the unified client and the C ABI.
 - The bundled WebSocket server acknowledges stream requests with ThetaData's stream-verification values: a successful subscribe, unsubscribe, or stop now returns `SUBSCRIBED` instead of `OK`, and a subscribe that arrives before streaming has started returns `ERROR` with a descriptive message instead of a false-positive `OK` that claimed success while installing nothing.
+- Oversized streaming credentials now fail with a typed invalid-parameter configuration error before the connection is opened instead of panicking the caller. The credentials payload is validated against the 255-byte protocol frame limit up front, and the error names both the limit and the actual size so a too-long email or password is reported as a normal recoverable failure.
+- Active streaming subscriptions are de-duplicated by contract and by full-stream security type, so a repeated subscribe no longer accumulates duplicate tracked entries that get replayed multiple times after a reconnect; unsubscribe still removes the tracked entry. The control channel that carries subscribe / unsubscribe / heartbeat commands to the I/O worker is now bounded, so a control-plane burst cannot grow it without limit: the heartbeat takes natural backpressure and the public subscribe / unsubscribe methods return a typed queue-full error rather than dropping a command or accumulating unbounded memory.
 
 ### Security
 

crates/thetadatadx/benches/bench_protocol.rs

@@ -74,7 +74,7 @@ fn bench_contract_roundtrip(c: &mut Criterion) {
 fn bench_build_credentials_payload(c: &mut Criterion) {
     c.bench_function("build_credentials_payload", |b| {
         b.iter(|| {
-            black_box(build_credentials_payload(
+            let _ = black_box(build_credentials_payload(
                 black_box("trader@example.com"),
                 black_box("s3cret_p4ssw0rd!"),
             ));

crates/thetadatadx/build_support_bin/sdk_surface/typescript.rs

@@ -47,7 +47,7 @@ fn ts_streaming_method(method: &MethodSpec) -> String {
                  the stream.\n\
                  \n\
                  Backpressure: a slow callback causes incoming events\n\
-                 to queue and, once the buffer is full, the oldest\n\
+                 to queue and, once the buffer is full, newly arriving\n\
                  events are dropped. The dropped count is observable\n\
                  via `droppedEventCount()`. The receive path is never\n\
                  blocked by a slow callback, so the upstream connection\n\

crates/thetadatadx/src/flatfiles/session.rs

@@ -100,7 +100,7 @@ pub(crate) async fn login(
     creds: &Credentials,
 ) -> Result<String, Error> {
     // CREDENTIALS frame.
-    let creds_payload = build_credentials_payload(&creds.email, creds.password());
+    let creds_payload = build_credentials_payload(&creds.email, creds.password())?;
     write_frame(stream, msg::CREDENTIALS, -1, &creds_payload).await?;
 
     // VERSION frame.
@@ -208,7 +208,7 @@ mod tests {
     #[test]
     fn credentials_payload_layout_is_stable() {
         // Verifies leading byte is 0x00 and userlen is BE u16.
-        let p = build_credentials_payload("a@b.c", "pw");
+        let p = build_credentials_payload("a@b.c", "pw").expect("valid creds");
         assert_eq!(p[0], 0x00);
         assert_eq!(&p[1..3], &5u16.to_be_bytes());
         assert_eq!(&p[3..8], b"a@b.c");

crates/thetadatadx/src/fpss/io_loop/mod.rs

@@ -810,7 +810,16 @@ where
         };
 
         // Re-authenticate on the new stream.
-        let cred_payload = build_credentials_payload(&creds.email, &creds.password);
+        let cred_payload = match build_credentials_payload(&creds.email, &creds.password) {
+            Ok(p) => p,
+            Err(e) => {
+                // Oversized credentials are a fatal configuration error, not a
+                // transient I/O fault: retrying cannot make them fit. Surface
+                // it and abandon the reconnect loop rather than spinning.
+                tracing::error!(error = %e, "credentials payload invalid; aborting reconnect");
+                break 'session;
+            }
+        };
         let frame = Frame::new(StreamMsgType::Credentials, cred_payload);
         if let Err(e) = write_frame(&mut new_stream, &frame) {
             tracing::warn!(error = %e, "failed to send credentials on reconnect");

crates/thetadatadx/src/fpss/io_loop/ping.rs

@@ -28,7 +28,7 @@ use super::super::protocol::build_ping_payload;
 // Reason: all parameters are moved into this function from a spawned thread closure.
 #[allow(clippy::needless_pass_by_value)]
 pub(in crate::fpss) fn ping_loop(
-    cmd_tx: std_mpsc::Sender<IoCommand>,
+    cmd_tx: std_mpsc::SyncSender<IoCommand>,
     shutdown: Arc<AtomicBool>,
     authenticated: Arc<AtomicBool>,
     interval: Duration,
@@ -55,6 +55,9 @@ pub(in crate::fpss) fn ping_loop(
             code: StreamMsgType::Ping,
             payload: ping_payload.clone(),
         };
+        // Blocking send on the bounded channel: the heartbeat takes natural
+        // backpressure if the I/O thread is momentarily behind, rather than
+        // dropping a ping. `send` only errors once the receiver hangs up.
         if cmd_tx.send(cmd).is_err() {
             // I/O thread has exited
             break;
@@ -88,7 +91,8 @@ mod tests {
     ///   -> Sleep cadence visible on `cmd_rx`.
     #[test]
     fn ping_loop_honors_configured_interval() {
-        let (cmd_tx, cmd_rx) = std_mpsc::channel::<IoCommand>();
+        let (cmd_tx, cmd_rx) =
+            std_mpsc::sync_channel::<IoCommand>(crate::fpss::CMD_CHANNEL_CAPACITY);
         let shutdown = Arc::new(AtomicBool::new(false));
         let authenticated = Arc::new(AtomicBool::new(true));
         let interval = Duration::from_millis(30);
@@ -143,7 +147,8 @@ mod tests {
     /// produces noticeably-spaced pings rather than the 100 ms default.
     #[test]
     fn ping_loop_with_longer_interval_paces_slower_than_default() {
-        let (cmd_tx, cmd_rx) = std_mpsc::channel::<IoCommand>();
+        let (cmd_tx, cmd_rx) =
+            std_mpsc::sync_channel::<IoCommand>(crate::fpss::CMD_CHANNEL_CAPACITY);
         let shutdown = Arc::new(AtomicBool::new(false));
         let authenticated = Arc::new(AtomicBool::new(true));
         let shutdown_clone = Arc::clone(&shutdown);