core: Rename methods to get start and end block for amp subgraphs

Author: lutter

core/src/amp_subgraph/runner/context.rs

@@ -77,7 +77,7 @@ impl<AC> Context<AC> {
             .map(|block_ptr| (block_ptr.number.compat(), block_ptr.hash.compat()))
     }
 
-    pub(super) fn min_start_block(&self) -> BlockNumber {
+    pub(super) fn start_block(&self) -> BlockNumber {
         self.manifest
             .data_sources
             .iter()
@@ -86,7 +86,7 @@ impl<AC> Context<AC> {
             .unwrap()
     }
 
-    pub(super) fn max_end_block(&self) -> BlockNumber {
+    pub(super) fn end_block(&self) -> BlockNumber {
         self.manifest
             .data_sources
             .iter()

core/src/amp_subgraph/runner/data_stream.rs

@@ -40,11 +40,11 @@ where
     debug!(logger, "Creating data stream";
         "from_block" => cx.latest_synced_block().unwrap_or(BlockNumber::MIN),
         "to_block" => latest_block,
-        "start_block" => cx.min_start_block(),
+        "start_block" => cx.start_block(),
         "max_block_range" => max_block_range,
     );
 
-    // State: (latest_queried_block, max_end_block, is_first)
+    // State: (latest_queried_block, end_block, is_first)
     let initial_state = (cx.latest_synced_block(), BlockNumber::MIN, true);
 
     stream::unfold(
@@ -132,12 +132,12 @@ fn build_data_stream<E>(
     table_ptrs: Arc<[TablePtr]>,
     buffer_size: usize,
     stopwatch: &StopwatchMetrics,
-    min_start_block: BlockNumber,
+    start_block: BlockNumber,
 ) -> BoxStream<'static, Result<(RecordBatchGroups, Arc<[TablePtr]>), Error>>
 where
     E: std::error::Error + IsDeterministic + Send + Sync + 'static,
 {
-    let mut min_start_block_checked = false;
+    let mut start_block_checked = false;
     let mut load_first_record_batch_group_section =
         Some(stopwatch.start_section("load_first_record_batch_group"));
 
@@ -151,14 +151,14 @@ where
 
             match result {
                 Ok(response) => {
-                    if !min_start_block_checked {
+                    if !start_block_checked {
                         if let Some(((first_block, _), _)) = response.0.first_key_value() {
-                            if *first_block < min_start_block {
+                            if *first_block < start_block {
                                 return Err(Error::NonDeterministic(anyhow!("chain reorg")));
                             }
                         }
 
-                        min_start_block_checked = true;
+                        start_block_checked = true;
                     }
 
                     Ok(response)

core/src/amp_subgraph/runner/mod.rs

@@ -101,7 +101,7 @@ where
 
         cx.metrics
             .deployment_target
-            .update(latest_block.min(cx.max_end_block()));
+            .update(latest_block.min(cx.end_block()));
 
         let mut deployment_is_failed = cx.store.health().await?.is_failed();
         let mut entity_cache = EntityCache::new(cx.store.cheap_clone());
@@ -131,12 +131,12 @@ where
         // source's endBlock. This handles the case where endBlock has no entity
         // data — the persisted block pointer never advances to endBlock, but the
         // server's latest block confirms all queries have been served.
-        if latest_block >= cx.max_end_block() {
+        if latest_block >= cx.end_block() {
             cx.metrics.deployment_synced.record(true);
 
             debug!(cx.logger, "Indexing completed; endBlock reached via server latest block";
                 "latest_block" => latest_block,
-                "max_end_block" => cx.max_end_block()
+                "end_block" => cx.end_block()
             );
             return Ok(());
         }