From cfe96be0a70683a2c00016a2357a633bd6586815 Mon Sep 17 00:00:00 2001
From: mehmetkr-31 <mehmet.kar@std.yildiz.edu.tr>
Date: Thu, 6 Aug 2026 20:29:46 +0300
Subject: [PATCH] fix(spammer): timestamp catch-up blocks individually

`catch_up_scan` propagated the `received_at` of the notification that
triggered the scan to every block it fetched, so all N blocks recovered
after a subscriber reconnect shared one observation time.

The latency tracker uses `BlockEvent::received_at` as the finalization
observation time for every transaction in the block, so a gap covering N
blocks reported N identical latencies anchored at the reconnect moment.
That both flattens the latency distribution and skews it: blocks fetched
late in the scan are credited with an observation time from before they
were fetched.

Capture `timestamp_now()` per block inside the scan loop instead. The
notification path is unchanged and still uses the header-arrival time.

Fixes #128

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 crates/spammer/src/latency/block_stream.rs | 17 +++++++++++------
 1 file changed, 11 insertions(+), 6 deletions(-)

diff --git a/crates/spammer/src/latency/block_stream.rs b/crates/spammer/src/latency/block_stream.rs
index 2c04fd43..8d2dcdf4 100644
--- a/crates/spammer/src/latency/block_stream.rs
+++ b/crates/spammer/src/latency/block_stream.rs
@@ -68,7 +68,9 @@ struct HeaderNotification {
 /// header-arrival timestamp for latency measurement.
 pub(super) struct BlockEvent {
     pub block: RpcBlock,
-    /// Propagated from [`HeaderNotification::received_at`].
+    /// Wall-clock time at which this block was observed: propagated from
+    /// [`HeaderNotification::received_at`] for notified blocks, or captured at
+    /// fetch time for blocks recovered by a catch-up scan.
     pub received_at: u64,
 }
 
@@ -306,7 +308,6 @@ impl BlockStream {
                     &mut ws_client,
                     *next_expected_height,
                     height - 1,
-                    notification.received_at,
                     block_sender,
                 )
                 .await?;
@@ -345,13 +346,16 @@ impl BlockStream {
     /// Fetch blocks from `start_height` through `end_height`
     /// (inclusive) and send them as [`BlockEvent`]s.
     ///
+    /// Each block is timestamped with `timestamp_now()` at the moment it is
+    /// fetched, so a scan covering many heights does not collapse into a
+    /// single observation time.
+    ///
     /// Returns `Err` if a block cannot be fetched after retries, so
     /// the caller can rebuild the client and retry the scan.
     async fn catch_up_scan(
         ws_client: &mut WsClient,
         start_height: u64,
         end_height: u64,
-        received_at: u64,
         block_sender: &Sender<BlockEvent>,
     ) -> Result<()> {
         if start_height > end_height {
@@ -365,10 +369,11 @@ impl BlockStream {
             let hex_height = format!("0x{height:x}");
             match Self::fetch_block(ws_client, ETH_GET_BLOCK_BY_NUMBER, &hex_height).await {
                 FetchResult::Ok(block) => {
-                    // Send the block to the tracker.
-                    // The timestamp for catch-up blocks is the
-                    // arrival time of the first notification.
+                    // Send the block to the tracker, timestamped at the moment
+                    // it was actually observed rather than at the arrival time
+                    // of the notification that triggered the scan.
                     // TODO: should we use the blocks' timestamps?
+                    let received_at = timestamp_now();
                     if block_sender
                         .send(BlockEvent { block, received_at })
                         .await