From 988a0e1d2372a555239fa92f5dcea11f5f6fbef8 Mon Sep 17 00:00:00 2001 From: Sonic Shih Date: Sun, 26 Jul 2026 12:15:29 +0800 Subject: [PATCH 1/2] fix(collector): acknowledge buffered proof rotations --- .../collector/src/bin/binance-lob-archiver.rs | 64 ++++++++----------- 1 file changed, 28 insertions(+), 36 deletions(-) diff --git a/rust_hft/tools/collector/src/bin/binance-lob-archiver.rs b/rust_hft/tools/collector/src/bin/binance-lob-archiver.rs index 93824935e..d6705b53e 100644 --- a/rust_hft/tools/collector/src/bin/binance-lob-archiver.rs +++ b/rust_hft/tools/collector/src/bin/binance-lob-archiver.rs @@ -2008,19 +2008,19 @@ async fn receive_url( } }, ProducerWait::PauseRequested => { + if let Some(exit) = acknowledge_rotation_pause( + producer_id, + &sender, + &mut rotation_pause, + &mut rotation_resume, + &mut last_pause_epoch, + &mut shutdown, + ) + .await? + { + return Ok(exit); + } if proof_events.is_empty() { - if let Some(exit) = acknowledge_rotation_pause( - producer_id, - &sender, - &mut rotation_pause, - &mut rotation_resume, - &mut last_pause_epoch, - &mut shutdown, - ) - .await? - { - return Ok(exit); - } subscription_proof_deadline = tokio::time::Instant::now() + subscription_proof_timeout; } @@ -4179,7 +4179,7 @@ mod tests { } #[tokio::test] - async fn buffered_reconnect_proof_event_drains_before_rotation_barrier() { + async fn buffered_reconnect_proof_event_acknowledges_rotation() { let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); let event_time_ms = now_ns().unwrap() / 1_000_000; @@ -4277,12 +4277,16 @@ mod tests { market_frame_rx.await.unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; pause_tx.send(1).unwrap(); - assert!( - tokio::time::timeout(Duration::from_millis(100), receiver.recv()) + assert!(matches!( + tokio::time::timeout(Duration::from_secs(1), receiver.recv()) .await - .is_err(), - "rotation barrier acknowledged before the buffered proof event was verified" - ); + .unwrap(), + Some(Event::RotationBarrier { + producer_id: 0, + epoch: 1 + }) + )); + resume_tx.send(1).unwrap(); proof_tx.send(()).unwrap(); assert!(matches!( receiver.recv().await, @@ -4292,14 +4296,6 @@ mod tests { receiver.recv().await, Some(Event::AggregateTrade { .. }) )); - assert!(matches!( - receiver.recv().await, - Some(Event::RotationBarrier { - producer_id: 0, - epoch: 1 - }) - )); - resume_tx.send(1).unwrap(); shutdown_tx.send(true).unwrap(); assert!(matches!( task.await.unwrap().unwrap(), @@ -4397,26 +4393,22 @@ mod tests { market_frame_rx.await.unwrap(); tokio::time::sleep(Duration::from_millis(100)).await; pause_tx.send(1).unwrap(); - assert!( - tokio::time::timeout(Duration::from_millis(100), receiver.recv()) - .await - .is_err(), - "rotation barrier acknowledged before the buffered proof event timed out" - ); assert!(matches!( tokio::time::timeout(Duration::from_secs(1), receiver.recv()) .await .unwrap(), - Some(Event::StreamDisconnected { .. }) - )); - assert!(matches!( - receiver.recv().await, Some(Event::RotationBarrier { producer_id: 0, epoch: 1 }) )); resume_tx.send(1).unwrap(); + assert!(matches!( + tokio::time::timeout(Duration::from_secs(1), receiver.recv()) + .await + .unwrap(), + Some(Event::StreamDisconnected { .. }) + )); shutdown_tx.send(true).unwrap(); assert!(matches!( task.await.unwrap().unwrap(), From acebaee8974174ee2398ba574f7436292ab028fc Mon Sep 17 00:00:00 2001 From: Sonic Shih Date: Sun, 26 Jul 2026 12:25:32 +0800 Subject: [PATCH 2/2] fix(collector): abandon buffered proof across rotation --- .../collector/src/bin/binance-lob-archiver.rs | 22 ++++++++++--------- 1 file changed, 12 insertions(+), 10 deletions(-) diff --git a/rust_hft/tools/collector/src/bin/binance-lob-archiver.rs b/rust_hft/tools/collector/src/bin/binance-lob-archiver.rs index d6705b53e..653545f9e 100644 --- a/rust_hft/tools/collector/src/bin/binance-lob-archiver.rs +++ b/rust_hft/tools/collector/src/bin/binance-lob-archiver.rs @@ -2008,6 +2008,7 @@ async fn receive_url( } }, ProducerWait::PauseRequested => { + let buffered_proof_interrupted = !proof_events.is_empty(); if let Some(exit) = acknowledge_rotation_pause( producer_id, &sender, @@ -2020,10 +2021,13 @@ async fn receive_url( { return Ok(exit); } - if proof_events.is_empty() { - subscription_proof_deadline = - tokio::time::Instant::now() + subscription_proof_timeout; + if buffered_proof_interrupted { + proof_failure = + Some("segment rotation interrupted buffered subscription proof".into()); + break; } + subscription_proof_deadline = + tokio::time::Instant::now() + subscription_proof_timeout; continue; } ProducerWait::Stopped => { @@ -4179,7 +4183,7 @@ mod tests { } #[tokio::test] - async fn buffered_reconnect_proof_event_acknowledges_rotation() { + async fn buffered_reconnect_proof_event_is_abandoned_after_rotation() { let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); let event_time_ms = now_ns().unwrap() / 1_000_000; @@ -4289,12 +4293,10 @@ mod tests { resume_tx.send(1).unwrap(); proof_tx.send(()).unwrap(); assert!(matches!( - receiver.recv().await, - Some(Event::StreamReconnected { .. }) - )); - assert!(matches!( - receiver.recv().await, - Some(Event::AggregateTrade { .. }) + tokio::time::timeout(Duration::from_secs(1), receiver.recv()) + .await + .unwrap(), + Some(Event::StreamDisconnected { .. }) )); shutdown_tx.send(true).unwrap(); assert!(matches!(