core, graph: Rename GRAPH_AMP_MAX_BLOCK_RANGE, lower default

Rename GRAPH_AMP_MAX_BLOCK_RANGE to just GRAPH_AMP_BLOCK_RANGE

Lower the default to 100k blocks from 2M blocks

Author: lutter

core/src/amp_subgraph/runner/context.rs

@@ -19,7 +19,7 @@ pub(in super::super) struct Context<AC> {
     pub(super) client: Arc<AC>,
     pub(super) store: Arc<dyn WritableStore>,
     pub(super) buffer_size: usize,
-    pub(super) max_block_range: usize,
+    pub(super) block_range: usize,
     pub(super) backoff: ExponentialBackoff,
     pub(super) deployment: DeploymentHash,
     pub(super) manifest: Manifest,
@@ -46,7 +46,7 @@ impl<AC> Context<AC> {
             client,
             store,
             buffer_size: env.buffer_size,
-            max_block_range: env.max_block_range,
+            block_range: env.block_range,
             backoff,
             deployment,
             manifest,

core/src/amp_subgraph/runner/data_stream.rs

@@ -34,14 +34,14 @@ where
     let client = cx.client.cheap_clone();
     let manifest = cx.manifest.clone();
     let buffer_size = cx.buffer_size;
-    let max_block_range = cx.max_block_range;
+    let block_range = cx.block_range;
     let stopwatch = cx.metrics.stopwatch.cheap_clone();
 
     debug!(logger, "Creating data stream";
         "from_block" => cx.latest_synced_block().unwrap_or(BlockNumber::MIN),
         "to_block" => latest_block,
         "start_block" => cx.start_block(),
-        "max_block_range" => max_block_range,
+        "block_range" => block_range,
     );
 
     // State: (latest_queried_block, end_block, is_first)
@@ -52,7 +52,7 @@ where
         move |(latest_queried_block, mut end_block, is_first)| {
             let block_ranges = next_block_ranges(
                 &manifest.data_sources,
-                max_block_range,
+                block_range,
                 latest_queried_block,
                 latest_block,
             );
@@ -171,21 +171,16 @@ where
 
 fn next_block_ranges(
     data_sources: &[DataSource],
-    max_block_range: usize,
+    block_range: usize,
     latest_queried_block: Option<BlockNumber>,
     latest_block: BlockNumber,
 ) -> HashMap<usize, RangeInclusive<BlockNumber>> {
     let block_ranges = data_sources
         .iter()
         .enumerate()
         .filter_map(|(i, data_source)| {
-            next_block_range(
-                max_block_range,
-                data_source,
-                latest_queried_block,
-                latest_block,
-            )
-            .map(|block_range| (i, block_range))
+            next_block_range(block_range, data_source, latest_queried_block, latest_block)
+                .map(|block_range| (i, block_range))
         })
         .collect::<HashMap<_, _>>();
 
@@ -204,7 +199,7 @@ fn next_block_ranges(
 }
 
 fn next_block_range(
-    max_block_range: usize,
+    block_range: usize,
     data_source: &DataSource,
     latest_queried_block: Option<BlockNumber>,
     latest_block: BlockNumber,
@@ -221,7 +216,7 @@ fn next_block_range(
     };
 
     let end_block = [
-        start_block.saturating_add(max_block_range as BlockNumber),
+        start_block.saturating_add(block_range as BlockNumber),
         data_source.source.end_block,
         latest_block,
     ]

docs/amp-powered-subgraphs.md

@@ -564,7 +564,7 @@ Amp-powered subgraphs feature introduces the following new ENV variables:
 - `GRAPH_AMP_FLIGHT_SERVICE_ADDRESS` – The address of the Amp Flight gRPC service. _Defaults to `None`, which disables support for Amp-powered subgraphs._
 - `GRAPH_AMP_FLIGHT_SERVICE_TOKEN` – Token used to authenticate Amp Flight gRPC service requests. _Defaults to `None`, which disables authentication._
 - `GRAPH_AMP_BUFFER_SIZE` – Maximum number of response batches to buffer in memory per stream for each SQL query. _Defaults to `1,000`._
-- `GRAPH_AMP_MAX_BLOCK_RANGE` – Maximum number of blocks to request per stream for each SQL query. _Defaults to `2,000,000`._
+- `GRAPH_AMP_BLOCK_RANGE` – Maximum number of blocks to request per stream for each SQL query. _Defaults to `100,000`._
 - `GRAPH_AMP_QUERY_RETRY_MIN_DELAY_SECONDS` – Minimum time to wait before retrying a failed SQL query to the Amp server. _Defaults to `1` second._
 - `GRAPH_AMP_QUERY_RETRY_MAX_DELAY_SECONDS` – Maximum time to wait before retrying a failed SQL query to the Amp server. _Defaults to `600` seconds._
 

graph/src/env/amp.rs

@@ -12,8 +12,8 @@ pub struct AmpEnv {
     /// Maximum number of blocks to request per stream for each SQL query.
     /// Limiting this value reduces load on the Amp server when processing heavy queries.
     ///
-    /// Defaults to `2,000,000`.
-    pub max_block_range: usize,
+    /// Defaults to `100,000`.
+    pub block_range: usize,
 
     /// Minimum time to wait before retrying a failed SQL query to the Amp server.
     ///
@@ -33,7 +33,7 @@ pub struct AmpEnv {
 
 impl AmpEnv {
     const DEFAULT_BUFFER_SIZE: usize = 1_000;
-    const DEFAULT_MAX_BLOCK_RANGE: usize = 2_000_000;
+    const DEFAULT_BLOCK_RANGE: usize = 100_000;
     const DEFAULT_QUERY_RETRY_MIN_DELAY: Duration = Duration::from_secs(1);
     const DEFAULT_QUERY_RETRY_MAX_DELAY: Duration = Duration::from_secs(600);
 
@@ -48,15 +48,15 @@ impl AmpEnv {
                     Some(value)
                 })
                 .unwrap_or(Self::DEFAULT_BUFFER_SIZE),
-            max_block_range: raw_env
-                .amp_max_block_range
+            block_range: raw_env
+                .amp_block_range
                 .map(|mut value| {
                     if value == 0 {
                         value = usize::MAX;
                     }
                     value
                 })
-                .unwrap_or(Self::DEFAULT_MAX_BLOCK_RANGE),
+                .unwrap_or(Self::DEFAULT_BLOCK_RANGE),
             query_retry_min_delay: raw_env
                 .amp_query_retry_min_delay_seconds
                 .map(Duration::from_secs)

graph/src/env/mod.rs

@@ -610,8 +610,8 @@ struct Inner {
 
     #[envconfig(from = "GRAPH_AMP_BUFFER_SIZE")]
     amp_buffer_size: Option<usize>,
-    #[envconfig(from = "GRAPH_AMP_MAX_BLOCK_RANGE")]
-    amp_max_block_range: Option<usize>,
+    #[envconfig(from = "GRAPH_AMP_BLOCK_RANGE")]
+    amp_block_range: Option<usize>,
     #[envconfig(from = "GRAPH_AMP_QUERY_RETRY_MIN_DELAY_SECONDS")]
     amp_query_retry_min_delay_seconds: Option<u64>,
     #[envconfig(from = "GRAPH_AMP_QUERY_RETRY_MAX_DELAY_SECONDS")]