diff --git a/backend/src/lib.rs b/backend/src/lib.rs index 76bcc06..5000cde 100644 --- a/backend/src/lib.rs +++ b/backend/src/lib.rs @@ -13,3 +13,6 @@ pub mod server { pub mod top_k_tracker { include!("lib/top_k_tracker.rs"); } +pub mod txn_hash_tracker { + include!("lib/txn_hash_tracker.rs"); +} diff --git a/backend/src/lib/server.rs b/backend/src/lib/server.rs index a3724bb..8a06703 100644 --- a/backend/src/lib/server.rs +++ b/backend/src/lib/server.rs @@ -21,6 +21,7 @@ use serde::{Deserialize, Serialize}; use crate::event_filter::{is_restricted_mode, load_restricted_filters}; use crate::event_listener::EventName; use crate::top_k_tracker::{AccessEntry, TopKTracker}; +use crate::txn_hash_tracker::TxnHashTracker; use super::event_filter::EventFilter; use super::event_listener::EventData; @@ -214,7 +215,7 @@ async fn run_event_forwarder_task( let mut accesses_reset_interval = tokio::time::interval(std::time::Duration::from_mins(5)); // Track current transaction hash per txn_idx - let mut current_txn_hashes: Vec> = vec![None; 10_000]; + let mut current_txn_hashes = TxnHashTracker::new(); let mut tps_tracker = TPSTracker::new(); @@ -237,7 +238,7 @@ async fn run_event_forwarder_task( // Track txn_hash from TxnHeaderStart events if let EventName::TxnHeaderStart = event_data.event_name { if let ExecEvent::TxnHeaderStart { txn_index, txn_header_start, .. } = &event_data.payload { - current_txn_hashes[*txn_index] = Some(txn_header_start.txn_hash.bytes); + current_txn_hashes.record(*txn_index, txn_header_start.txn_hash.bytes); } else { unreachable!(); } @@ -245,8 +246,8 @@ async fn run_event_forwarder_task( // Populate txn_hash for events that have txn_idx if let Some(txn_idx) = event_data.txn_idx { - if let Some(Some(hash)) = current_txn_hashes.get(txn_idx) { - event_data.txn_hash = Some(*hash); + if let Some(hash) = current_txn_hashes.get(txn_idx) { + event_data.txn_hash = Some(hash); } } @@ -255,13 +256,17 @@ async fn run_event_forwarder_task( match event_data.event_name { EventName::BlockStart => { tps_event = Some(EventDataOrMetrics::TPS(tps_tracker.get_tps())); + // txn_idx is scoped to a single block; drop anything left over + // (e.g. from a TxnEnd missed due to an event-ring gap) so it + // can't accumulate for the life of the process. + current_txn_hashes.reset(); } EventName::TxnHeaderStart => { tps_tracker.record_tx(); } EventName::TxnEnd => { if let Some(txn_idx) = event_data.txn_idx { - current_txn_hashes[txn_idx] = None; + current_txn_hashes.clear(txn_idx); } } EventName::AccountAccess => { diff --git a/backend/src/lib/txn_hash_tracker.rs b/backend/src/lib/txn_hash_tracker.rs new file mode 100644 index 0000000..01c5de6 --- /dev/null +++ b/backend/src/lib/txn_hash_tracker.rs @@ -0,0 +1,118 @@ +use std::collections::HashMap; + +/// Tracks the in-flight transaction hash for each `txn_idx` within the current block. +/// +/// Backed by a map rather than a fixed-size buffer: `txn_idx` comes straight off the +/// event ring with no upper bound, so a fixed-capacity buffer indexed directly by +/// `txn_idx` would panic on any block with more transactions than the buffer's capacity. +#[derive(Default)] +pub struct TxnHashTracker { + hashes: HashMap, +} + +impl TxnHashTracker { + pub fn new() -> Self { + Self::default() + } + + /// Record the hash for a transaction that just started. + pub fn record(&mut self, txn_idx: usize, hash: [u8; 32]) { + self.hashes.insert(txn_idx, hash); + } + + /// Look up the hash for a transaction, if one is currently tracked. + pub fn get(&self, txn_idx: usize) -> Option<[u8; 32]> { + self.hashes.get(&txn_idx).copied() + } + + /// Stop tracking a transaction once it has ended. + pub fn clear(&mut self, txn_idx: usize) { + self.hashes.remove(&txn_idx); + } + + /// Drop all tracked hashes. + /// + /// `txn_idx` is scoped to a single block, so nothing left over from a + /// previous block is ever valid to keep. Call this on `BlockStart` so a + /// missed `TxnEnd` (e.g. from an event-ring gap) can't leave an orphaned + /// entry retained for the forwarder's lifetime. + pub fn reset(&mut self) { + self.hashes.clear(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn record_then_get_returns_the_hash() { + let mut tracker = TxnHashTracker::new(); + let hash = [7u8; 32]; + + tracker.record(3, hash); + + assert_eq!(tracker.get(3), Some(hash)); + } + + #[test] + fn get_on_unknown_index_returns_none() { + let tracker = TxnHashTracker::new(); + + assert_eq!(tracker.get(3), None); + } + + #[test] + fn clear_removes_the_tracked_hash() { + let mut tracker = TxnHashTracker::new(); + tracker.record(3, [7u8; 32]); + + tracker.clear(3); + + assert_eq!(tracker.get(3), None); + } + + /// The bug this struct fixes: the old code stored hashes in a `Vec` fixed at + /// 10_000 entries and indexed it directly with `txn_idx`, which panics on any + /// index at or beyond that capacity. A `HashMap` has no such ceiling. + #[test] + fn handles_txn_idx_far_beyond_the_old_fixed_capacity_of_10_000() { + let mut tracker = TxnHashTracker::new(); + let hash = [9u8; 32]; + let large_idx = 50_000; + + tracker.record(large_idx, hash); + + assert_eq!(tracker.get(large_idx), Some(hash)); + + tracker.clear(large_idx); + assert_eq!(tracker.get(large_idx), None); + } + + #[test] + fn re_recording_the_same_index_overwrites_the_previous_hash() { + let mut tracker = TxnHashTracker::new(); + tracker.record(1, [1u8; 32]); + tracker.record(1, [2u8; 32]); + + assert_eq!(tracker.get(1), Some([2u8; 32])); + } + + /// Covers the orphaned-entry case: a `TxnEnd` can be missed (e.g. the event-ring + /// reader hits a gap and resets), leaving `clear` never called for that txn_idx. + /// `reset` is the backstop that bounds memory regardless, by dropping everything + /// at the start of the next block. + #[test] + fn reset_drops_entries_never_cleared_by_a_missed_txn_end() { + let mut tracker = TxnHashTracker::new(); + tracker.record(1, [1u8; 32]); + tracker.record(2, [2u8; 32]); + // txn_idx 2's TxnEnd is "missed" - no clear(2) call. + tracker.clear(1); + + tracker.reset(); + + assert_eq!(tracker.get(1), None); + assert_eq!(tracker.get(2), None); + } +}