chore(sequencer): Renaming round buffer to ring buffer + provider.get_latest_values only available for test

Author: HristoStaykov

apps/sequencer/src/aggregate_batch_consensus_processor.rs

@@ -61,7 +61,7 @@ pub async fn aggregation_batch_consensus_loop(
                             let providers = sequencer_state.providers.read().await;
                             let mut provider = providers.get(net.as_str()).unwrap().lock().await;
                             let ids_vec: Vec<_> = t.updated_feeds_ids.iter().copied().collect();
-                            warn!("Tiemed out batch {t:?} while collectiong reporters' signatures for net {net}. Decreasing the round buffer indices for feed_ids: {ids_vec:?}");
+                            warn!("Tiemed out batch {t:?} while collectiong reporters' signatures for net {net}. Decreasing the ring buffer indices for feed_ids: {ids_vec:?}");
                             decrement_feed_rb_indices(&ids_vec, net.as_str(), &mut provider).await
                         }
 

apps/sequencer/src/providers/eth_send_utils.rs

@@ -29,7 +29,7 @@ use crate::{
     sequencer_state::SequencerState,
 };
 use blocksense_feeds_processing::adfs_gen_calldata::{
-    adfs_serialize_updates, get_neighbour_feed_ids, RoundBufferIndices,
+    adfs_serialize_updates, get_neighbour_feed_ids, RingBufferIndices,
 };
 use blocksense_metrics::{
     dec_metric, inc_metric, inc_vec_metric,
@@ -559,7 +559,7 @@ pub async fn eth_batch_send_to_contract(
 
                     // If the nonce in the contract increased and the next state root hash is not as we expect,
                     // another sequencer was able to post updates for the current block height before this one.
-                    // We need to take this into account and reread the round counters of the feeds.
+                    // We need to take this into account and reread the ring buffer indices of the feeds.
                     info!("Updates to contract already posted, network {net}, block_height {block_height}, latest_nonce {latest_nonce}, previous_nonce {nonce}, merkle_root in contract {prev_calldata_merkle_tree_root:?}");
                     try_to_sync(
                         net.as_str(),
@@ -836,7 +836,7 @@ pub async fn eth_batch_send_to_contract(
         provider.calldata_merkle_tree_frontier = next_calldata_merkle_tree;
         provider.merkle_root_in_contract = None;
         debug!("Successfully updated contract in network `{net}` block height {block_height} Merkle root {root:?}");
-    } // TODO: Reread round counters + latest state hash from contract
+    } // TODO: Reread ring buffer indices + latest state hash from contract
     drop(provider);
     debug!("Released a read/write lock on provider state in network `{net}` block height {block_height}");
 
@@ -1391,14 +1391,14 @@ pub async fn eth_batch_send_to_all_contracts(
 async fn log_rb_indices(
     prefix: &str,
     updated_feeds: &Vec<EncodedFeedId>,
-    rb_indices: &mut RoundBufferIndices,
+    rb_indices: &mut RingBufferIndices,
     net: &str,
 ) {
     let mut debug_string =
         format!("{prefix} for net = {net} and updated_feeds = {updated_feeds:?} ");
     for feed in updated_feeds {
-        let round_index = rb_indices.get(feed).unwrap_or(&0);
-        debug_string.push_str(format!("{feed} = {round_index}; ").as_str());
+        let ring_buffer_index = rb_indices.get(feed).unwrap_or(&0);
+        debug_string.push_str(format!("{feed} = {ring_buffer_index}; ").as_str());
     }
     debug!(debug_string);
 }
@@ -1417,8 +1417,8 @@ pub async fn increment_feeds_rb_indices(
     .await;
 
     for feed in updated_feeds {
-        let round_buffer_index = provider.rb_indices.entry(*feed).or_insert(0);
-        *round_buffer_index += 1;
+        let ring_buffer_index = provider.rb_indices.entry(*feed).or_insert(0);
+        *ring_buffer_index += 1;
     }
 
     log_rb_indices(
@@ -1429,8 +1429,8 @@ pub async fn increment_feeds_rb_indices(
     )
     .await;
 }
-// Since we update the round buffer index when we post the tx and before we
-// receive its receipt if the tx fails we need to decrease the round indices.
+// Since we update the ring buffer index when we post the tx and before we
+// receive its receipt if the tx fails we need to decrease the ring buffer indices.
 pub async fn decrement_feed_rb_indices(
     updated_feeds: &Vec<EncodedFeedId>,
     net: &str,
@@ -1445,9 +1445,9 @@ pub async fn decrement_feed_rb_indices(
     .await;
 
     for feed in updated_feeds {
-        let round_buffer_index = provider.rb_indices.entry(*feed).or_insert(0);
-        if *round_buffer_index > 0 {
-            *round_buffer_index -= 1;
+        let ring_buffer_index = provider.rb_indices.entry(*feed).or_insert(0);
+        if *ring_buffer_index > 0 {
+            *ring_buffer_index -= 1;
         }
     }
 
@@ -1575,7 +1575,7 @@ mod tests {
     ) {
         let key1 = EncodedFeedId::new(0x1F as FeedId, 0);
         let key2 = EncodedFeedId::new(0x0FFF as FeedId, 0);
-        let mut rb_indices = RoundBufferIndices::new();
+        let mut rb_indices = RingBufferIndices::new();
         rb_indices.insert(key1, 7);
         rb_indices.insert(key2, 8);
         let mut strides_and_decimals = HashMap::new();

apps/sequencer/src/providers/provider.rs

@@ -14,7 +14,7 @@ use alloy::{
 use alloy_primitives::{keccak256, B256, U256};
 use alloy_u256_literal::u256;
 use blocksense_feeds_processing::adfs_gen_calldata::{
-    calc_row_index, RoundBufferIndices, MAX_HISTORY_ELEMENTS_PER_FEED,
+    calc_row_index, RingBufferIndices, MAX_HISTORY_ELEMENTS_PER_FEED,
     NUM_FEED_IDS_IN_RB_INDEX_RECORD,
 };
 use blocksense_utils::{EncodedFeedId, FeedId};
@@ -32,7 +32,7 @@ use blocksense_data_feeds::feeds_processing::{
 use blocksense_feed_registry::registry::FeedAggregateHistory;
 use blocksense_feed_registry::types::FeedType;
 use blocksense_metrics::{metrics::ProviderMetrics, process_provider_getter};
-use eyre::{eyre, Result};
+use eyre::{bail, eyre, Result};
 use paste::paste;
 use ringbuf::traits::{Consumer, Observer};
 use serde::{Deserialize, Serialize};
@@ -123,7 +123,7 @@ pub struct RpcProvider {
     pub feeds_variants: HashMap<EncodedFeedId, FeedVariant>,
     pub contracts: Vec<Contract>,
     pub rpc_url: Url,
-    pub rb_indices: RoundBufferIndices,
+    pub rb_indices: RingBufferIndices,
     num_tx_in_progress: u32,
     pub inflight: InflightObservations,
 }
@@ -241,18 +241,18 @@ async fn load_data_from_chain(
         let res = provider.load_rb_indices_from_chain(&feeds_config).await;
         match res {
             Ok(mut rb_indices) => {
-                info!("Loaded round buffer indices from chain {network} = {rb_indices:?}");
+                info!("Loaded ring buffer indices from chain {network} = {rb_indices:?}");
                 for (_id, counter) in rb_indices.iter_mut() {
                     *counter = (*counter + 1) % MAX_HISTORY_ELEMENTS_PER_FEED;
                 }
                 provider.rb_indices = rb_indices;
             }
             Err(err) => {
-                error!("Error when loading round buffer indices for {network} = {err}");
+                error!("Error when loading ring buffer indices for {network} = {err}");
             }
         }
     } else {
-        warn!("Skipping loading round buffer indices from chain {network}");
+        warn!("Skipping loading ring buffer indices from chain {network}");
     }
 }
 
@@ -356,7 +356,7 @@ impl RpcProvider {
             feeds_variants,
             contracts,
             rpc_url,
-            rb_indices: RoundBufferIndices::new(),
+            rb_indices: RingBufferIndices::new(),
             num_tx_in_progress: 0,
             inflight: InflightObservations::new(),
         }
@@ -574,6 +574,7 @@ impl RpcProvider {
         r
     }
 
+    #[cfg(test)]
     pub async fn get_latest_values(
         &self,
         encoded_feed_ids: &[EncodedFeedId],
@@ -584,20 +585,16 @@ impl RpcProvider {
         let mut variants: HashMap<EncodedFeedId, FeedVariant> = HashMap::new();
         for encoded_feed_id in encoded_feed_ids.iter() {
             let Some(variant) = self.feeds_variants.get(encoded_feed_id) else {
-                return Err(eyre!(
-                    "Unknown variant and number of digits for feed with encoded_feed_id = {encoded_feed_id}"
-                ));
+                bail!("Unknown variant and number of digits for feed with encoded_feed_id = {encoded_feed_id}");
             };
             variants.insert(*encoded_feed_id, variant.clone());
         }
 
         for encoded_feed_id in encoded_feed_ids {
             let Some(feed_variant) = variants.get(encoded_feed_id) else {
-                return Err(eyre!(
-                    "Unknown variant and number of digits for feed with id (logical error) = {encoded_feed_id}"
-                ));
+                bail!("Unknown variant and number of digits for feed with id (logical error) = {encoded_feed_id}");
             };
-            // abi.encodePacked(bytes1(0x82), stride, uint120(id))
+            // abi.encodePacked(bytes1(0x83), stride, uint120(id))
             let calldata = DynSolValue::Tuple(vec![
                 DynSolValue::Uint(U256::from(0x83_u8), 8),
                 DynSolValue::Uint(U256::from(feed_variant.stride), 8),
@@ -697,9 +694,7 @@ impl RpcProvider {
             Ok(res) => res,
             Err(e) => {
                 warn!("Timed out on get_tx_retry_params while deploying contract {contract_name} in network `{network}`: {e}!");
-                return Err(eyre!(
-                    "failed to get_tx_retry_params for network `{network}"
-                ));
+                bail!("failed to get_tx_retry_params for network `{network}");
             }
         };
 
@@ -1638,7 +1633,7 @@ mod tests {
 
                 let prov = sequencer_state.providers.read().await;
                 let p = prov.get(network).unwrap();
-                let round = p
+                let rb_index = p
                     .lock()
                     .await
                     .get_latest_rb_index(&encoded_feed_id)
@@ -1654,10 +1649,10 @@ mod tests {
                     .as_ref()
                     .expect("Expected correct value in contract for feed id {feed_id}");
 
-                // Assert that the value of the round counter in the contract is as expected.
+                // Assert that the value of the rb_index counter in the contract is as expected.
                 // Note: The sequencer tracks the index of the *next* slot to write,
                 // while the contract stores the index of the *last* written value.
-                assert_eq!(round.index, wrapped_val);
+                assert_eq!(rb_index.index, wrapped_val);
 
                 assert_eq!(val.value, new_update);
 

apps/sequencer/src/providers/reorg_tracking.rs

@@ -398,7 +398,7 @@ impl ReorgTracker {
                         };
                     };
 
-                    // 3) If we detected divergence, resync round-buffer indices from chain
+                    // 3) If we detected divergence, resync ring buffer indices from chain
                     if need_resync_indices {
                         let mut provider = provider_mutex.lock().await;
                         if let Some(contract) = provider.get_latest_contract() {

libs/feeds_processing/src/adfs_gen_calldata.rs

@@ -14,7 +14,7 @@ use once_cell::sync::Lazy;
 pub const MAX_HISTORY_ELEMENTS_PER_FEED: u64 = 8192;
 pub const NUM_FEED_IDS_IN_RB_INDEX_RECORD: u128 = 16;
 
-pub type RoundBufferIndices = HashMap<EncodedFeedId, u64>; // for each key (feed_id) we store its round buffer index
+pub type RingBufferIndices = HashMap<EncodedFeedId, u64>; // for each key (feed_id) we store its ring buffer index
 
 static STRIDES_SIZES: Lazy<HashMap<u8, u32>> = Lazy::new(|| {
     let mut map = HashMap::new(); // TODO: confirm the correct values for the strides we will support
@@ -67,9 +67,9 @@ pub fn calc_row_index(feed_id: FeedId, stride: Stride) -> alloy_primitives::Uint
 pub async fn adfs_serialize_updates(
     net: &str,
     feed_updates: &BatchedAggregatesToSend,
-    rb_indices: Option<&RoundBufferIndices>,
+    rb_indices: Option<&RingBufferIndices>,
     strides_and_decimals: HashMap<EncodedFeedId, FeedStrideAndDecimals>,
-    feeds_rb_indexes: &mut HashMap<EncodedFeedId, u64>, /* The round buffer indices table for the relevant feeds. If the rb_indices are provided,
+    feeds_rb_indexes: &mut HashMap<EncodedFeedId, u64>, /* The ring buffer indices table for the relevant feeds. If the rb_indices are provided,
                                                         this map will be filled with the update count for each feed from it. If the
                                                         rb_indices is None, feeds_rb_indexes will be used as the source of the updates
                                                         count. */
@@ -179,8 +179,8 @@ pub async fn adfs_serialize_updates(
         result.append(&mut result_bytes);
     }
 
-    // In case feed_metrics is none, the feeds_rb_indexes contains all the round indices needed for serialization.
-    // We use them to populate feeds_info map based on which the round indices will be serialized
+    // In case feed_metrics is none, the feeds_rb_indexes contains all the ring buffer indices needed for serialization.
+    // We use them to populate feeds_info map based on which the ring buffer indices will be serialized
     if rb_indices.is_none() {
         for (feed_id, rb_index) in feeds_rb_indexes.iter() {
             if let Some(strides_and_decimals) = strides_and_decimals.get(feed_id) {
@@ -306,7 +306,7 @@ pub mod tests {
         config_init: HashMap<EncodedFeedId, FeedStrideAndDecimals>,
     ) -> (
         BatchedAggregatesToSend,
-        RoundBufferIndices,
+        RingBufferIndices,
         HashMap<EncodedFeedId, FeedStrideAndDecimals>,
     ) {
         let updates = BatchedAggregatesToSend {
@@ -317,9 +317,9 @@ pub mod tests {
                 .collect(),
         };
 
-        let mut rb_indices = RoundBufferIndices::new();
-        for (feed_id, round) in rb_indices_init.iter() {
-            rb_indices.insert(*feed_id, *round);
+        let mut rb_indices = RingBufferIndices::new();
+        for (feed_id, rb_index) in rb_indices_init.iter() {
+            rb_indices.insert(*feed_id, *rb_index);
         }
 
         (updates, rb_indices, config_init)
@@ -340,7 +340,7 @@ pub mod tests {
             (encoded_feed_id_for_stride_zero(4), "4890"),
             (encoded_feed_id_for_stride_zero(5), "5abc"),
         ];
-        let round_counters_init = vec![
+        let rb_indices_init = vec![
             (encoded_feed_id_for_stride_zero(1), 6),
             (encoded_feed_id_for_stride_zero(2), 5),
             (encoded_feed_id_for_stride_zero(3), 4),
@@ -350,7 +350,7 @@ pub mod tests {
 
         let config_init = default_config();
         let (updates, rb_indices, config) =
-            setup_updates_rb_indices_and_config(&updates_init, &round_counters_init, config_init);
+            setup_updates_rb_indices_and_config(&updates_init, &rb_indices_init, config_init);
 
         let expected_result = "000000050102400c0107123432676435730002400501022456000260040102367800028003010248900002a00201025abc010000000000000500040003000200000000000000000000000000000000000000000e80000000000000000000000000000000000600000000000000000000000000000000000000000000000000000000";
 
@@ -394,7 +394,7 @@ pub mod tests {
             (encoded_feed_id_for_stride_zero(4), "4890"),
             (encoded_feed_id_for_stride_zero(5), "5abc"),
         ];
-        let round_counters_init = vec![
+        let rb_indices_init = vec![
             (encoded_feed_id_for_stride_zero(1), 6),
             (encoded_feed_id_for_stride_zero(2), 5),
             (encoded_feed_id_for_stride_zero(3), 4),
@@ -404,7 +404,7 @@ pub mod tests {
 
         let config_init = default_config();
         let (updates, mut rb_indices, config) =
-            setup_updates_rb_indices_and_config(&updates_init, &round_counters_init, config_init);
+            setup_updates_rb_indices_and_config(&updates_init, &rb_indices_init, config_init);
         rb_indices.insert(encoded_feed_id_for_stride_zero(6), 5);
 
         let expected_result = "000000050102400c0107123432676435730002400501022456000260040102367800028003010248900002a00201025abc010000000000000500040003000200040000000000000000000000000000000000000e80000000000000000000000000000000000600000000000000000000000000000000000000000000000000000000";
@@ -449,7 +449,7 @@ pub mod tests {
             (encoded_feed_id_for_stride_zero(4), "4890"),
             (encoded_feed_id_for_stride_zero(5), "5abc"),
         ];
-        let round_counters_init = vec![
+        let rb_indices_init = vec![
             (encoded_feed_id_for_stride_zero(1), 6),
             (encoded_feed_id_for_stride_zero(2), 5),
             (encoded_feed_id_for_stride_zero(3), 4),
@@ -458,13 +458,13 @@ pub mod tests {
         ];
 
         let config_init = default_config();
-        let (updates, mut round_counters, config) =
-            setup_updates_rb_indices_and_config(&updates_init, &round_counters_init, config_init);
-        round_counters.insert(encoded_feed_id_for_stride_zero(6), 5);
+        let (updates, mut rb_indices, config) =
+            setup_updates_rb_indices_and_config(&updates_init, &rb_indices_init, config_init);
+        rb_indices.insert(encoded_feed_id_for_stride_zero(6), 5);
 
         let expected_result = "000000050102400c0107123432676435730002400501022456000260040102367800028328010248900002a00201025abc010000000000000500040328000200040000000000000000000000000000000000000e80000000000000000000000000000000000600000000000000000000000000000000000000000000000000000000";
 
-        let mut feeds_rounds = HashMap::new();
+        let mut batch_rb_indices = HashMap::new();
 
         // Call as it will be in the sequencer
         assert_eq!(
@@ -473,20 +473,20 @@ pub mod tests {
                 adfs_serialize_updates(
                     net,
                     &updates,
-                    Some(&round_counters),
+                    Some(&rb_indices),
                     config.clone(),
-                    &mut feeds_rounds,
+                    &mut batch_rb_indices,
                 )
                 .await
                 .unwrap()
             )
         );
 
-        // Call as it will be in the reporter (feeds_rounds provided by the sequencer)
+        // Call as it will be in the reporter (rb_indices provided by the sequencer)
         assert_eq!(
             expected_result,
             hex::encode(
-                adfs_serialize_updates(net, &updates, None, config, &mut feeds_rounds,)
+                adfs_serialize_updates(net, &updates, None, config, &mut rb_indices,)
                     .await
                     .unwrap()
             )

libs/feeds_processing/src/utils.rs

@@ -422,7 +422,7 @@ pub async fn validate(
 pub mod tests {
     use blocksense_utils::FeedId;
 
-    use crate::adfs_gen_calldata::RoundBufferIndices;
+    use crate::adfs_gen_calldata::RingBufferIndices;
     use std::collections::HashSet;
 
     use super::*;
@@ -507,7 +507,7 @@ pub mod tests {
                 end_slot_timestamp: 1677654323,
             },
         ];
-        let mut feeds_rb_indices: RoundBufferIndices = HashMap::new();
+        let mut feeds_rb_indices: RingBufferIndices = HashMap::new();
         feeds_rb_indices.insert(EncodedFeedId::new(1, 0), 1000);
         feeds_rb_indices.insert(EncodedFeedId::new(5, 0), 2000);
         feeds_rb_indices.insert(EncodedFeedId::new(11, 0), 3000);