From 7fb583e027d26845af17a0b119f1c7c513d3d675 Mon Sep 17 00:00:00 2001 From: preview Date: Wed, 17 Jun 2026 23:13:25 +0200 Subject: [PATCH] fix(fpss): correct derived-bar accounting and escalate write failures to reconnect Out-of-order trade corrections (nonzero records_back) reprice a print already counted in the running bar. The derived OHLCVC accumulator dropped that field, so every correction re-added its size and trade, double-counting volume and count on the derive path. Thread records_back into the accumulator and let a correction adjust only the price extremes and the close, never volume or count. Steady-state and post-reconnect command-drain write failures were logged and swallowed, leaving the loop reading on a socket whose write side had broken and deferring recovery to the next read timeout while the queued subscribe or command was lost. Escalate a write error to an immediate reconnect, mirroring the replay-flush path: the in-session drain publishes Disconnected and breaks to the reconnect decision, and the post-reconnect drain re-enters the session loop. The login handshake looped on read with no wall-clock bound and an unbounded pending-control buffer, so a peer streaming Connected/Ping/Restart without ever sending Metadata reset the per-read timeout on every frame and grew the buffer without limit, wedging the reconnect thread. Bound the handshake with a wall-clock deadline derived from the read timeout and cap the buffered control frames, returning a timeout or protocol error past those limits. Co-Authored-By: Claude Opus 4.8 --- crates/thetadatadx/src/fpss/accumulator.rs | 92 ++++++++-- crates/thetadatadx/src/fpss/decode.rs | 19 ++- crates/thetadatadx/src/fpss/io_loop/login.rs | 171 +++++++++++++++++-- crates/thetadatadx/src/fpss/io_loop/mod.rs | 41 ++++- crates/thetadatadx/src/fpss/mod.rs | 2 +- 5 files changed, 294 insertions(+), 31 deletions(-) diff --git a/crates/thetadatadx/src/fpss/accumulator.rs b/crates/thetadatadx/src/fpss/accumulator.rs index 09433fea0..63b99dc0d 100644 --- a/crates/thetadatadx/src/fpss/accumulator.rs +++ b/crates/thetadatadx/src/fpss/accumulator.rs @@ -74,7 +74,27 @@ impl OhlcvcAccumulator { size: i32, price_type: i32, date: i32, + records_back: i32, ) { + // A nonzero `records_back` marks an out-of-order correction: the print + // refers to a trade already reported `records_back` records earlier in + // the stream, so its volume and count are already inside the running + // bar. Re-adding `size`/`count` would double-count the correction, so + // a correction only adjusts the price extremes (and the close) without + // advancing volume or count. A correction never opens a fresh bar: it + // cannot legitimately reference a print on a date the accumulator has + // not seen, so it is folded into the current bar when initialized. + if self.initialized && date == self.date && records_back != 0 { + let adjusted_price = change_price_type(price, price_type, self.price_type); + if adjusted_price > self.high { + self.high = adjusted_price; + } + if adjusted_price < self.low { + self.low = adjusted_price; + } + self.close = adjusted_price; + return; + } // A trade on a new date opens a fresh bar. Without this, an already- // initialized accumulator merges the new date's trade onto the prior // date's open/high/low and cumulative volume/count while the emitted @@ -162,7 +182,7 @@ mod tests { fn ohlcvc_accumulator_first_trade_initializes() { let mut acc = OhlcvcAccumulator::new(); assert!(!acc.initialized); - acc.process_trade(34200000, 15025, 100, 8, 20240315); + acc.process_trade(34200000, 15025, 100, 8, 20240315, 0); assert!(acc.initialized); assert_eq!(acc.open, 15025); assert_eq!(acc.high, 15025); @@ -175,9 +195,9 @@ mod tests { #[test] fn ohlcvc_accumulator_updates() { let mut acc = OhlcvcAccumulator::new(); - acc.process_trade(34200000, 15025, 100, 8, 20240315); - acc.process_trade(34200100, 15100, 200, 8, 20240315); - acc.process_trade(34200200, 14950, 50, 8, 20240315); + acc.process_trade(34200000, 15025, 100, 8, 20240315, 0); + acc.process_trade(34200100, 15100, 200, 8, 20240315, 0); + acc.process_trade(34200200, 14950, 50, 8, 20240315, 0); assert_eq!(acc.open, 15025); assert_eq!(acc.high, 15100); assert_eq!(acc.low, 14950); @@ -192,7 +212,7 @@ mod tests { acc.init_from_server( 34200000, 15000, 15100, 14900, 15050, 1000_i64, 10_i64, 8, 20240315, ); - acc.process_trade(34200300, 15200, 50, 8, 20240315); + acc.process_trade(34200300, 15200, 50, 8, 20240315, 0); assert_eq!(acc.high, 15200); assert_eq!(acc.low, 14900); assert_eq!(acc.volume, 1050); @@ -202,8 +222,8 @@ mod tests { #[test] fn ohlcvc_accumulator_no_overflow_on_high_volume() { let mut acc = OhlcvcAccumulator::new(); - acc.process_trade(34200000, 15025, i32::MAX, 8, 20240315); - acc.process_trade(34200100, 15100, i32::MAX, 8, 20240315); + acc.process_trade(34200000, 15025, i32::MAX, 8, 20240315, 0); + acc.process_trade(34200100, 15100, i32::MAX, 8, 20240315, 0); // Would overflow i32 (2 * 2_147_483_647 = 4_294_967_294), fine in i64 assert_eq!(acc.volume, 2 * i64::from(i32::MAX)); assert_eq!(acc.count, 2); @@ -238,8 +258,8 @@ mod tests { fn ohlcvc_accumulator_rolls_on_date_change() { let mut acc = OhlcvcAccumulator::new(); // Day 1: two trades accumulate. - acc.process_trade(57_600_000, 15_025, 100, 8, 20240315); - acc.process_trade(57_600_100, 15_100, 200, 8, 20240315); + acc.process_trade(57_600_000, 15_025, 100, 8, 20240315, 0); + acc.process_trade(57_600_100, 15_100, 200, 8, 20240315, 0); assert_eq!(acc.date, 20240315); assert_eq!(acc.volume, 300); assert_eq!(acc.count, 2); @@ -247,7 +267,7 @@ mod tests { // Day 2: the first trade on the new date opens a fresh bar. Open, // high, low, close all reset to this trade's price; volume and count // restart from this trade alone; the date advances. - acc.process_trade(34_200_000, 14_950, 50, 8, 20240318); + acc.process_trade(34_200_000, 14_950, 50, 8, 20240318, 0); assert_eq!(acc.date, 20240318); assert_eq!(acc.open, 14_950); assert_eq!(acc.high, 14_950); @@ -258,6 +278,54 @@ mod tests { assert_eq!(acc.ms_of_day, 34_200_000); } + /// An out-of-order correction (`records_back != 0`) reprices a print that + /// is already inside the running bar. Its size and trade are already part + /// of `volume`/`count`, so a correction must NOT advance either; doing so + /// double-counts the print. The correction may still pull the price + /// extremes and the close, since the corrected price can be a new high, + /// low, or last. + #[test] + fn ohlcvc_accumulator_correction_does_not_double_count() { + let mut acc = OhlcvcAccumulator::new(); + acc.process_trade(34_200_000, 15_025, 100, 8, 20240315, 0); + acc.process_trade(34_200_100, 15_100, 200, 8, 20240315, 0); + assert_eq!(acc.volume, 300); + assert_eq!(acc.count, 2); + + // A correction one record back at a new extreme: volume/count are + // unchanged, but high and close track the corrected price. + acc.process_trade(34_200_200, 15_200, 200, 8, 20240315, 1); + assert_eq!(acc.volume, 300, "correction must not re-add volume"); + assert_eq!(acc.count, 2, "correction must not re-add count"); + assert_eq!(acc.high, 15_200, "corrected price can set a new high"); + assert_eq!(acc.close, 15_200, "corrected price updates the close"); + assert_eq!(acc.low, 15_025); + + // A subsequent fresh print (records_back == 0) resumes accumulation + // from the un-corrupted running totals. + acc.process_trade(34_200_300, 14_900, 50, 8, 20240315, 0); + assert_eq!( + acc.volume, 350, + "fresh print after a correction adds normally" + ); + assert_eq!(acc.count, 3); + assert_eq!(acc.low, 14_900); + } + + /// A correction folds into an already-open bar even when it carries a low + /// price: it adjusts the low without resetting open/volume/count, so a + /// correction can never masquerade as a bar reset. + #[test] + fn ohlcvc_accumulator_correction_adjusts_low_without_reset() { + let mut acc = OhlcvcAccumulator::new(); + acc.process_trade(34_200_000, 15_025, 100, 8, 20240315, 0); + acc.process_trade(34_200_100, 15_100, 200, 8, 20240315, 2); + assert_eq!(acc.open, 15_025, "correction must not reset the open"); + assert_eq!(acc.high, 15_100); + assert_eq!(acc.volume, 100, "correction must not add to volume"); + assert_eq!(acc.count, 1, "correction must not add to count"); + } + #[test] fn change_price_type_tests() { assert_eq!(change_price_type(15025, 8, 8), 15025); @@ -305,12 +373,12 @@ mod tests { #[test] fn ohlcvc_accumulator_rescale_matches_java_wrap() { let mut acc = OhlcvcAccumulator::new(); - acc.process_trade(34200000, 71_396_865, 1, 4, 20240315); + acc.process_trade(34200000, 71_396_865, 1, 4, 20240315, 0); assert_eq!(acc.price_type, 4); assert_eq!(acc.high, 71_396_865); // Tick at price_type=8 needs *10^4 to rescale into pricetype=4 and // wraps to +1_004_078_864, which exceeds the prior high. - acc.process_trade(34200100, 71_396_865, 1, 8, 20240315); + acc.process_trade(34200100, 71_396_865, 1, 8, 20240315, 0); assert_eq!(acc.high, 1_004_078_864); assert_eq!(acc.close, 1_004_078_864); // Low stays at the original (smaller) value. diff --git a/crates/thetadatadx/src/fpss/decode.rs b/crates/thetadatadx/src/fpss/decode.rs index dfcbcb821..9c582e177 100644 --- a/crates/thetadatadx/src/fpss/decode.rs +++ b/crates/thetadatadx/src/fpss/decode.rs @@ -491,11 +491,13 @@ pub fn decode_frame( }) }; - // Extract for OHLCVC derivation (format-aware). - let (ms_of_day, size, price, price_type, date) = if n_data <= 8 { - (buf[0], buf[2], buf[4], buf[6], buf[7]) + // Extract for OHLCVC derivation (format-aware). The + // 8-field layout carries no `records_back`, so every + // 8-field print is a fresh trade (`records_back == 0`). + let (ms_of_day, size, price, price_type, date, records_back) = if n_data <= 8 { + (buf[0], buf[2], buf[4], buf[6], buf[7], 0) } else { - (buf[0], buf[7], buf[9], buf[14], buf[15]) + (buf[0], buf[7], buf[9], buf[14], buf[15], buf[13]) }; // Derive OHLCVC from trade only if enabled and the @@ -503,7 +505,14 @@ pub fn decode_frame( let ohlcvc_event = if derive_ohlcvc { if let Some(acc) = delta_state.ohlcvc.get_mut(&contract_id) { if acc.initialized { - acc.process_trade(ms_of_day, price, size, price_type, date); + acc.process_trade( + ms_of_day, + price, + size, + price_type, + date, + records_back, + ); let apt = acc.price_type; match ( strict_fpss_price(acc.open, apt), diff --git a/crates/thetadatadx/src/fpss/io_loop/login.rs b/crates/thetadatadx/src/fpss/io_loop/login.rs index 9666c740f..ba36ac8c1 100644 --- a/crates/thetadatadx/src/fpss/io_loop/login.rs +++ b/crates/thetadatadx/src/fpss/io_loop/login.rs @@ -7,6 +7,8 @@ //! captured for replay onto the event bus once the Disruptor producer //! is live. +use std::time::{Duration, Instant}; + use crate::tdbe::types::enums::{RemoveReason, StreamMsgType}; use crate::error::Error; @@ -22,6 +24,25 @@ pub(in crate::fpss) enum LoginResult { Disconnected(RemoveReason), } +/// Upper bound on the number of pre-`Metadata` control frames buffered during +/// a single handshake. A legitimate handshake emits a handful (`Connected`, +/// an early `Ping`, a `Restart`); a peer that streams control frames without +/// ever sending `Metadata` would otherwise grow `pending_control` without +/// limit. Past this count the handshake is treated as a protocol violation. +const MAX_HANDSHAKE_PENDING_CONTROL: usize = 64; + +/// Wall-clock budget for a single handshake, independent of the per-read +/// socket timeout. A chatty peer that sends a control frame just before each +/// read timeout would otherwise reset the per-read deadline forever; this +/// caps the total time the reconnect thread can spend in one handshake. +fn handshake_deadline(read_timeout: Duration) -> Instant { + // Generous relative to a single read so a legitimately slow but + // progressing handshake is never cut off, yet bounded so a chatty + // no-`Metadata` peer cannot wedge the reconnect thread. + let budget = (read_timeout.saturating_mul(5)).max(Duration::from_secs(15)); + Instant::now() + budget +} + /// Wait for the server's login response (blocking). /// /// Reads frames until METADATA or DISCONNECTED. @@ -42,19 +63,57 @@ pub(in crate::fpss) enum LoginResult { pub(in crate::fpss) fn wait_for_login( stream: &mut connection::FpssStream, pending_control: &mut Vec, + read_timeout: Duration, ) -> Result { - wait_for_login_generic(stream, pending_control) + let deadline = handshake_deadline(read_timeout); + wait_for_login_generic( + stream, + pending_control, + deadline, + MAX_HANDSHAKE_PENDING_CONTROL, + Instant::now, + ) } /// Read-generic variant of [`wait_for_login`] for unit-testable handshake /// coverage. Holds the full dispatch logic so both the TLS-backed entry /// point above and in-memory test harnesses can drive it against a /// buffer of pre-canned frames. -fn wait_for_login_generic( +/// +/// `deadline` is a wall-clock backstop and `max_pending_control` caps the +/// number of pre-`Metadata` control frames. Together they bound a peer that +/// streams control frames without ever completing the handshake: the +/// per-read socket timeout alone cannot, because each frame resets it. +/// `now` injects the clock so tests can drive the deadline deterministically. +fn wait_for_login_generic( stream: &mut R, pending_control: &mut Vec, -) -> Result { + deadline: Instant, + max_pending_control: usize, + mut now: C, +) -> Result +where + R: std::io::Read, + C: FnMut() -> Instant, +{ loop { + if now() >= deadline { + return Err(Error::Fpss { + kind: crate::error::FpssErrorKind::Timeout, + message: "login handshake did not complete within the handshake deadline" + .to_string(), + }); + } + + if pending_control.len() >= max_pending_control { + return Err(Error::Fpss { + kind: crate::error::FpssErrorKind::ProtocolError, + message: format!( + "login handshake buffered {max_pending_control} control frames without METADATA" + ), + }); + } + let frame = read_frame(stream)?.ok_or_else(|| Error::Fpss { kind: crate::error::FpssErrorKind::Disconnected, message: "connection closed during login handshake".to_string(), @@ -135,6 +194,22 @@ mod tests { v } + /// Drive the handshake with bounds that never trip for the + /// dispatch/ordering tests: a far-future deadline and the production cap. + fn run_handshake( + stream: &mut R, + pending_control: &mut Vec, + ) -> Result { + let deadline = Instant::now() + Duration::from_secs(3_600); + wait_for_login_generic( + stream, + pending_control, + deadline, + MAX_HANDSHAKE_PENDING_CONTROL, + Instant::now, + ) + } + /// A CONNECTED frame arriving BEFORE METADATA must be captured in /// `pending_control` so the io_loop can forward the buffered /// `StreamControl::Connected` to the event bus. Without this capture @@ -148,7 +223,7 @@ mod tests { let mut cursor = std::io::Cursor::new(buf); let mut pending: Vec = Vec::new(); - let result = wait_for_login_generic(&mut cursor, &mut pending) + let result = run_handshake(&mut cursor, &mut pending) .expect("wait_for_login_generic must succeed when Metadata arrives"); match result { LoginResult::Success(p) => assert_eq!(p, "test-perms"), @@ -170,7 +245,7 @@ mod tests { let mut cursor = std::io::Cursor::new(buf); let mut pending: Vec = Vec::new(); - let result = wait_for_login_generic(&mut cursor, &mut pending) + let result = run_handshake(&mut cursor, &mut pending) .expect("wait_for_login_generic must succeed when Metadata arrives"); assert!(matches!(result, LoginResult::Success(_))); assert!( @@ -194,7 +269,7 @@ mod tests { let mut cursor = std::io::Cursor::new(buf); let mut pending: Vec = Vec::new(); - let result = wait_for_login_generic(&mut cursor, &mut pending) + let result = run_handshake(&mut cursor, &mut pending) .expect("Disconnected frame must produce LoginResult::Disconnected, not Err"); assert!(matches!(result, LoginResult::Disconnected(_))); assert!(pending.is_empty()); @@ -265,7 +340,7 @@ mod tests { let mut cursor = std::io::Cursor::new(buf); let mut pending: Vec = Vec::new(); - let result = wait_for_login_generic(&mut cursor, &mut pending) + let result = run_handshake(&mut cursor, &mut pending) .expect("wait_for_login_generic must succeed when Metadata arrives"); assert!(matches!(result, LoginResult::Success(_))); assert_eq!(pending.len(), 1, "PING must surface as a typed control"); @@ -293,7 +368,7 @@ mod tests { let mut cursor = std::io::Cursor::new(buf); let mut pending: Vec = Vec::new(); - let result = wait_for_login_generic(&mut cursor, &mut pending) + let result = run_handshake(&mut cursor, &mut pending) .expect("wait_for_login_generic must succeed when Metadata arrives"); assert!(matches!(result, LoginResult::Success(_))); assert_eq!(pending.len(), 1); @@ -310,7 +385,7 @@ mod tests { let mut cursor = std::io::Cursor::new(buf); let mut pending: Vec = Vec::new(); - let result = wait_for_login_generic(&mut cursor, &mut pending) + let result = run_handshake(&mut cursor, &mut pending) .expect("wait_for_login_generic must succeed when Metadata arrives"); assert!(matches!(result, LoginResult::Success(_))); assert_eq!(pending.len(), 1); @@ -330,7 +405,7 @@ mod tests { let mut cursor = std::io::Cursor::new(buf); let mut pending: Vec = Vec::new(); - let result = wait_for_login_generic(&mut cursor, &mut pending) + let result = run_handshake(&mut cursor, &mut pending) .expect("wait_for_login_generic must succeed when Metadata arrives"); assert!(matches!(result, LoginResult::Success(_))); assert_eq!(pending.len(), 4); @@ -339,4 +414,80 @@ mod tests { assert!(matches!(pending[2], StreamControl::ReconnectedServer)); assert!(matches!(pending[3], StreamControl::Restart)); } + + /// A peer that streams control frames without ever sending METADATA must + /// not wedge the handshake. With the per-read socket timeout reset by + /// every frame, the wall-clock deadline is the only backstop: once it + /// passes, the handshake returns a timeout instead of looping forever. + #[test] + fn wait_for_login_chatty_no_metadata_hits_deadline() { + // A long buffer of PING frames and no METADATA: read_frame always has + // a frame to return, so only the deadline can stop the loop. + let mut buf = Vec::new(); + for _ in 0..1_000 { + buf.extend_from_slice(&wire_frame(StreamMsgType::Ping, &[0x00])); + } + let mut cursor = std::io::Cursor::new(buf); + + // A clock that jumps past the deadline after a few iterations, + // emulating wall-clock time advancing while the peer keeps talking. + let start = Instant::now(); + let deadline = start + Duration::from_secs(10); + let mut ticks = 0u32; + let clock = move || { + ticks += 1; + if ticks > 3 { + deadline + Duration::from_secs(1) + } else { + start + } + }; + + let mut pending: Vec = Vec::new(); + let result = wait_for_login_generic( + &mut cursor, + &mut pending, + deadline, + MAX_HANDSHAKE_PENDING_CONTROL, + clock, + ); + match result { + Err(Error::Fpss { kind, .. }) => { + assert!(matches!(kind, crate::error::FpssErrorKind::Timeout)); + } + Err(other) => panic!("expected an Fpss timeout error, got {other:?}"), + Ok(_) => panic!("a chatty no-METADATA peer must trip the handshake deadline"), + } + } + + /// A peer that floods control frames without METADATA must not grow + /// `pending_control` without bound: once the cap is reached the handshake + /// returns a protocol error rather than buffering forever. + #[test] + fn wait_for_login_caps_pending_control() { + let cap = 8usize; + // More control frames than the cap, and no METADATA. + let mut buf = Vec::new(); + for _ in 0..(cap + 4) { + buf.extend_from_slice(&wire_frame(StreamMsgType::Connected, &[])); + } + let mut cursor = std::io::Cursor::new(buf); + + // Far-future deadline so the cap, not the clock, is what stops it. + let deadline = Instant::now() + Duration::from_secs(3_600); + let mut pending: Vec = Vec::new(); + let result = wait_for_login_generic(&mut cursor, &mut pending, deadline, cap, Instant::now); + match result { + Err(Error::Fpss { kind, .. }) => { + assert!(matches!(kind, crate::error::FpssErrorKind::ProtocolError)); + } + Err(other) => panic!("expected an Fpss protocol error, got {other:?}"), + Ok(_) => panic!("flooding control frames past the cap must error"), + } + assert_eq!( + pending.len(), + cap, + "buffering must stop exactly at the cap, not past it" + ); + } } diff --git a/crates/thetadatadx/src/fpss/io_loop/mod.rs b/crates/thetadatadx/src/fpss/io_loop/mod.rs index b79ec6b7f..172067b9f 100644 --- a/crates/thetadatadx/src/fpss/io_loop/mod.rs +++ b/crates/thetadatadx/src/fpss/io_loop/mod.rs @@ -727,7 +727,31 @@ where write_raw_frame_no_flush(writer, code, &payload) }; if let Err(e) = result { - tracing::warn!(error = %e, "failed to write frame"); + // A steady-state write failure means the socket's + // write side is broken. Reading on never recovers + // the queued subscribe/command that just failed, so + // escalate to reconnect immediately rather than + // deferring to the next read timeout. Mirror the + // disconnect-and-break shape the read-error and + // EOF branches use. + tracing::warn!(error = %e, "frame write failed; treating socket as broken and reconnecting"); + if producer + .try_publish(|slot| { + slot.event = + FpssEventInternal::Control(StreamControl::Disconnected { + reason: RemoveReason::Unspecified, + }); + }) + .is_err() + { + dropped.fetch_add(1, Ordering::Relaxed); + tracing::warn!( + target: "thetadatadx::fpss::io_loop", + "ring full while publishing Disconnected (write error); dropped", + ); + } + authenticated.store(false, Ordering::Release); + break 'inner RemoveReason::Unspecified; } } Ok(IoCommand::Shutdown) => { @@ -1029,7 +1053,11 @@ where } let mut reconnect_pending_control: Vec = Vec::new(); - let login_result = match wait_for_login(&mut new_stream, &mut reconnect_pending_control) { + let login_result = match wait_for_login( + &mut new_stream, + &mut reconnect_pending_control, + read_timeout, + ) { Ok(r) => r, Err(e) => { tracing::warn!(error = %e, "login failed on reconnect"); @@ -1320,7 +1348,14 @@ where write_raw_frame_no_flush(writer, code, &payload) }; if let Err(e) = result { - tracing::warn!(error = %e, "failed to write queued frame on reconnect"); + // The freshly reconnected socket is already broken if + // the queued-command drain cannot write. Re-enter the + // reconnect loop so recovery is driven the same way the + // re-subscribe replay above handles a mid-flush + // failure, rather than running a session whose queued + // command never reached the server. + tracing::warn!(error = %e, "queued-frame write failed on reconnect; treating socket as broken and reconnecting"); + continue 'session; } } Ok(IoCommand::Shutdown) => { diff --git a/crates/thetadatadx/src/fpss/mod.rs b/crates/thetadatadx/src/fpss/mod.rs index 78ab5d424..a04666412 100644 --- a/crates/thetadatadx/src/fpss/mod.rs +++ b/crates/thetadatadx/src/fpss/mod.rs @@ -1244,7 +1244,7 @@ impl StreamingClient { // the same sequence the post-METADATA `decode_frame` dispatch // emits. let mut pending_control: Vec = Vec::new(); - let login_result = wait_for_login(&mut stream, &mut pending_control)?; + let login_result = wait_for_login(&mut stream, &mut pending_control, read_timeout)?; let permissions = match login_result { LoginResult::Success(permissions) => {