fix(core, graph): minor adjustments after rebase

Author: isum

core/src/amp_subgraph/manager.rs

@@ -146,7 +146,7 @@ where
                         metrics,
                     );
 
-                    let runner_result = runner::new_runner(runner_context)(cancel_token).await;
+                    let runner_result = runner::new_runner(runner_context, cancel_token).await;
 
                     match manager.subgraph_store.stop_subgraph(&deployment).await {
                         Ok(()) => {

core/src/amp_subgraph/monitor.rs

@@ -2,9 +2,9 @@
 //!
 //! # Terminology used in this module
 //!
-//! `active subgraph` - A subgraph that was started and is still tracked.
+//! `active subgraph` - A subgraph that was started and is still kept in memory in the list of started subgraphs.
 //! `running subgraph` - A subgraph that has an instance that is making progress or stopping.
-//! `subgraph instance` - A background process that executes the subgraph runner future.
+//! `subgraph instance` - A background task that executes the subgraph runner future.
 
 use std::{
     collections::{hash_map::Entry, HashMap},
@@ -65,7 +65,7 @@ pub(super) struct Monitor {
     /// The channel that is used to send subgraph commands.
     ///
     /// Every subgraph start and stop request results in a command that is sent to the
-    /// background process that manages the subgraph instances.
+    /// background task that manages the subgraph instances.
     command_tx: mpsc::UnboundedSender<Command>,
 
     /// When a subgraph starts it is assigned a sequential ID.
@@ -87,10 +87,10 @@ pub(super) struct Monitor {
 impl Monitor {
     /// Creates a new subgraph monitor.
     ///
-    /// Spawns a background process that manages the subgraph start and stop requests.
+    /// Spawns a background task that manages the subgraph start and stop requests.
     ///
     /// A new cancel token is derived from the `cancel_token` and only the derived token is used by the
-    /// subgraph monitor and its background process.
+    /// subgraph monitor and its background task.
     pub(super) fn new(logger_factory: &LoggerFactory, cancel_token: &CancellationToken) -> Self {
         let logger = logger_factory.component_logger("AmpSubgraphMonitor", None);
         let logger_factory = Arc::new(logger_factory.with_parent(logger));
@@ -394,9 +394,9 @@ impl Monitor {
         }
     }
 
-    /// Spawns a background process that executes the subgraph runner future.
+    /// Spawns a background task that executes the subgraph runner future.
     ///
-    /// An additional background process is spawned to handle the graceful shutdown of the subgraph runner,
+    /// An additional background task is spawned to handle the graceful shutdown of the subgraph runner,
     /// and to ensure correct behaviour even if the subgraph runner panics.
     fn start_subgraph(
         logger: Logger,
@@ -502,7 +502,7 @@ impl Drop for Monitor {
     }
 }
 
-/// Represents a background process that executes the subgraph runner future.
+/// Represents a background task that executes the subgraph runner future.
 struct SubgraphInstance {
     id: u32,
     handle: JoinHandle<()>,

core/src/amp_subgraph/runner/compat.rs

@@ -1,3 +1,5 @@
+//! This is a temporary compatibility module until the graph-node is fully migrated to `alloy`.
+
 use alloy::primitives::{BlockHash, BlockNumber};
 use chrono::{DateTime, Utc};
 

core/src/amp_subgraph/runner/data_processing.rs

@@ -103,8 +103,14 @@ async fn process_record_batch_group<AC>(
         return Ok(entity_cache);
     }
 
-    let block_timestamp = decode_block_timestamp(&record_batches)
-        .map_err(|e| e.context("failed to decode block timestamp"))?;
+    let block_timestamp = if cx.manifest.schema.has_aggregations() {
+        decode_block_timestamp(&record_batches)
+            .map_err(|e| e.context("failed to decode block timestamp"))?
+    } else {
+        // TODO: Block timestamp is only required for subgraph aggregations.
+        //       Make it optional at the store level.
+        DateTime::<Utc>::MIN_UTC
+    };
 
     for record_batch in record_batches {
         let StreamRecordBatch {
@@ -231,6 +237,14 @@ async fn process_record_batch<AC>(
     Ok(())
 }
 
+/// Decodes the block timestamp from the first matching column in `record_batches`.
+///
+/// Iterates through the provided record batches and returns the timestamp from
+/// the first batch that contains a valid block timestamp column.
+///
+/// # Preconditions
+///
+/// All entries in `record_batches` must belong to the same record batch group.
 fn decode_block_timestamp(record_batches: &[StreamRecordBatch]) -> Result<DateTime<Utc>, Error> {
     let mut last_error: Option<Error> = None;
 

core/src/amp_subgraph/runner/mod.rs

@@ -9,7 +9,7 @@ mod reorg_handler;
 use std::time::{Duration, Instant};
 
 use anyhow::Result;
-use futures::{future::BoxFuture, StreamExt};
+use futures::StreamExt;
 use graph::{
     amp::Client, cheap_clone::CheapClone, components::store::EntityCache,
     data::subgraph::schema::SubgraphError,
@@ -24,48 +24,45 @@ use self::{
 
 pub(super) use self::context::Context;
 
-pub(super) fn new_runner<AC>(
+pub(super) async fn new_runner<AC>(
     mut cx: Context<AC>,
-) -> Box<dyn FnOnce(CancellationToken) -> BoxFuture<'static, Result<()>> + Send + 'static>
+    cancel_token: CancellationToken,
+) -> Result<()>
 where
     AC: Client + Send + Sync + 'static,
 {
-    Box::new(move |cancel_token| {
-        Box::pin(async move {
-            let indexing_duration_handle = tokio::spawn({
-                let mut instant = Instant::now();
-                let indexing_duration = cx.metrics.indexing_duration.clone();
-
-                async move {
-                    loop {
-                        tokio::time::sleep(Duration::from_secs(1)).await;
-
-                        let prev_instant = std::mem::replace(&mut instant, Instant::now());
-                        indexing_duration.record(prev_instant.elapsed());
-                    }
-                }
-            });
-
-            let result = cancel_token
-                .run_until_cancelled(run_indexing_with_retries(&mut cx))
-                .await;
+    let indexing_duration_handle = tokio::spawn({
+        let mut instant = Instant::now();
+        let indexing_duration = cx.metrics.indexing_duration.clone();
 
-            indexing_duration_handle.abort();
+        async move {
+            loop {
+                tokio::time::sleep(Duration::from_secs(1)).await;
 
-            match result {
-                Some(result) => result?,
-                None => {
-                    debug!(cx.logger, "Processed cancel signal");
-                }
+                let prev_instant = std::mem::replace(&mut instant, Instant::now());
+                indexing_duration.record(prev_instant.elapsed());
             }
+        }
+    });
+
+    let result = cancel_token
+        .run_until_cancelled(run_indexing_with_retries(&mut cx))
+        .await;
+
+    indexing_duration_handle.abort();
+
+    match result {
+        Some(result) => result?,
+        None => {
+            debug!(cx.logger, "Processed cancel signal");
+        }
+    }
 
-            cx.metrics.deployment_status.stopped();
+    cx.metrics.deployment_status.stopped();
 
-            debug!(cx.logger, "Waiting for the store to finish processing");
-            cx.store.flush().await?;
-            Ok(())
-        })
-    })
+    debug!(cx.logger, "Waiting for the store to finish processing");
+    cx.store.flush().await?;
+    Ok(())
 }
 
 async fn run_indexing<AC>(cx: &mut Context<AC>) -> Result<(), Error>