From bd30c11e79ca97b8b87d3fac5804fe2962ade35f Mon Sep 17 00:00:00 2001 From: userFRM Date: Thu, 30 Jul 2026 17:09:53 +0200 Subject: [PATCH] gateway: scan the inflated init content without putting it in front of the engine MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The logon ACK arrives DEFLATE-compressed inside `8=FIXCOMP` envelopes, and the routing tags the connect path needs are in the plaintext. It decompressed every segment and appended the result to `init_data` so the tag scan could see it — but `init_data` is also the buffer handed to the engine, which is given the burst still compressed and decompresses the same segments itself. So the hot loop read each message in the init burst twice from a single delivery: once out of the segment it inflated, once from the plaintext copy appended behind it. Every account record, resting order and bulletin in the burst was handled twice. The inflation is now taken into a copy. `init_scan_buffer` takes the burst by reference and returns the buffer to scan, so the engine's copy cannot be the one that grows — reintroducing this would take changing the signature, not adding a line. Nothing else changes: the tag scan sees exactly what it saw, the engine is handed exactly what arrived. Closes #317. --- src/gateway.rs | 110 +++++++++++++++++++++++++++++++++---------------- 1 file changed, 74 insertions(+), 36 deletions(-) diff --git a/src/gateway.rs b/src/gateway.rs index c238ee2d..ec370742 100644 --- a/src/gateway.rs +++ b/src/gateway.rs @@ -914,6 +914,48 @@ pub struct GatewayConfig { pub code_provider: Option, } +/// The buffer the init burst is scanned in. +/// +/// The auth-server's logon ACK arrives DEFLATE-compressed inside one or more +/// `8=FIXCOMP` envelopes (per ib-agent#129). The compressed body is ~30 kB on +/// the wire but expands to ~48 kB of plaintext carrying the routing tags +/// 6145/6171/8008, which the tag scan needs to see. +/// +/// The inflated plaintext belongs to that scan and nowhere else. The engine is +/// handed the burst exactly as it arrived and decompresses the same segments +/// itself, so appending the plaintext to the engine's copy delivered every +/// message in the burst twice from a single delivery — once from the segment, +/// once from the appended copy (ibx#317). Takes the burst by reference so the +/// engine's copy cannot be the one that grows. +fn init_scan_buffer(init_data: &[u8]) -> Vec { + let mut scan = init_data.to_vec(); + let mut cursor = 0usize; + while cursor + 12 < init_data.len() { + if init_data[cursor..].starts_with(b"8=FIXCOMP\x01") { + if let Some(total_len) = fixcomp::fixcomp_length(&init_data[cursor..]) { + let segment = &init_data[cursor..cursor + total_len.min(init_data.len() - cursor)]; + let inflated = fixcomp::fixcomp_decompress(segment).unwrap_or_else(|e| { + log::warn!("Init FIXCOMP segment at offset {}: dropping malformed frame: {}", cursor, e); + Vec::new() + }); + let inflated_bytes: usize = inflated.iter().map(|m| m.len() + 1).sum(); + log::info!( + "Init FIXCOMP segment at offset {}: {} compressed → {} inner messages, ~{} inflated bytes", + cursor, total_len, inflated.len(), inflated_bytes, + ); + for inner in inflated { + scan.extend_from_slice(&inner); + scan.push(b'\x01'); + } + cursor += total_len; + continue; + } + } + cursor += 1; + } + scan +} + impl Gateway { /// Connect to IB: auth + logon + data farm connections. /// Returns Gateway + farm Connection + auth Connection + optional historical data Connection. @@ -1405,44 +1447,10 @@ impl Gateway { init_data.len(), read_start.elapsed(), ); - // The auth-server's logon ACK arrives DEFLATE-compressed inside one or - // more `8=FIXCOMP` envelopes (per ib-agent#129). The compressed body - // is ~30 kB on the wire but expands to ~48 kB plaintext containing - // the routing tags 6145/6171/8008. Walk the buffer, decompress every - // FIXCOMP segment, and concatenate the plaintext with init_data so the - // existing tag-scan loop below sees the inflated content. - let mut inflated_extra: Vec = Vec::new(); - let mut cursor = 0usize; - while cursor + 12 < init_data.len() { - if init_data[cursor..].starts_with(b"8=FIXCOMP\x01") { - if let Some(total_len) = fixcomp::fixcomp_length(&init_data[cursor..]) { - let segment = &init_data[cursor..cursor + total_len.min(init_data.len() - cursor)]; - let inflated = fixcomp::fixcomp_decompress(segment).unwrap_or_else(|e| { - log::warn!("Init FIXCOMP segment at offset {}: dropping malformed frame: {}", cursor, e); - Vec::new() - }); - let inflated_bytes: usize = inflated.iter().map(|m| m.len() + 1).sum(); - log::info!( - "Init FIXCOMP segment at offset {}: {} compressed → {} inner messages, ~{} inflated bytes", - cursor, total_len, inflated.len(), inflated_bytes, - ); - for inner in inflated { - inflated_extra.extend_from_slice(&inner); - inflated_extra.push(b'\x01'); - } - cursor += total_len; - continue; - } - } - cursor += 1; - } - if !inflated_extra.is_empty() { - log::info!("Inflated {} bytes of FIXCOMP content; appending to scan buffer", inflated_extra.len()); - init_data.extend_from_slice(&inflated_extra); - } + let scan_data = init_scan_buffer(&init_data); // Scan init response for account ID and gateway-local init tags - let init_str = String::from_utf8_lossy(&init_data); + let init_str = String::from_utf8_lossy(&scan_data); // TEMP diagnostic (ib-agent#128 follow-up): log every part containing // "farm" or "hmds" so we can locate the routing tags. for part in init_str.split('\x01') { @@ -1885,6 +1893,36 @@ pub use crate::config::{chrono_free_timestamp, days_to_ymd}; mod tests { use super::*; + /// ibx#317: the init burst is handed to the engine still compressed, and + /// the engine decompresses the same segments itself. Appending the inflated + /// plaintext to that buffer — rather than to a copy taken for the local tag + /// scan — put every message in the burst in front of the engine twice from + /// a single delivery. + #[test] + fn the_inflated_init_content_is_scanned_but_not_handed_to_the_engine() { + let inner = b"8=FIX.4.2\x0135=B\x0158=ROUTING\x016145=farm-a\x0110=000\x01"; + let mut burst = b"8=FIX.4.2\x0135=A\x01".to_vec(); + burst.extend_from_slice(&fixcomp::fixcomp_build(inner)); + burst.extend_from_slice(b"8=FIX.4.2\x0135=0\x01"); + + fn count(haystack: &[u8], needle: &[u8]) -> usize { + haystack.windows(needle.len()).filter(|w| *w == needle).count() + } + + let before = count(&burst, b"58=ROUTING"); + let scan = init_scan_buffer(&burst); + + assert_eq!( + count(&scan, b"58=ROUTING"), before + 1, + "the tag scan gains exactly one inflated copy of the segment's content", + ); + assert!(scan.starts_with(&burst), "and still sees everything that arrived"); + assert_eq!( + count(&burst, b"58=ROUTING"), before, + "and the buffer the engine is handed is not the one that grew", + ); + } + #[test] fn token_short_hash_deterministic() { let token = BigUint::from(123456789u64);