graph, store: Change how the time of the last rollup is determined

When graph-node is restarted, we need to determine for subgraphs with
aggregations when the last rollup was triggered to ensure aggregations get
filled without gaps or duplication.

The code used the `block_time` column in the PoI table for that but that is
not correct as the PoI table only records blocks and times for which the
subgraph actually has writes. When the subgraph scans through a largish
number of blocks without changes, we only update the head pointer but also
do rollups as aggregation intervals pass. Because of that, we might perform
a rollup without a corresponding entry in the PoI table.

With this change, we actually find the maximum timestamp from all
aggregation tables to tell us when the last rollup was triggered as that
data reflects when rollups happened accurately.

For safety, this new behavior can be turned off by setting
`GRAPH_STORE_LAST_ROLLUP_FROM_POI=true` to return to the old buggy behavior
in case the new behavior causes some other unexpected problems.

Fixes #5530

Author: lutter

graph/src/env/store.rs

@@ -120,6 +120,14 @@ pub struct EnvVarsStore {
     pub use_brin_for_all_query_types: bool,
     /// Temporary env var to disable certain lookups in the chain store
     pub disable_block_cache_for_lookup: bool,
+    /// Temporary env var to fall back to the old broken way of determining
+    /// the time of the last rollup from the POI table instead of the new
+    /// way that fixes
+    /// https://github.com/graphprotocol/graph-node/issues/5530 Remove this
+    /// and all code that is dead as a consequence once this has been vetted
+    /// sufficiently, probably after 2024-12-01
+    /// Defaults to `false`, i.e. using the new fixed behavior
+    pub last_rollup_from_poi: bool,
 }
 
 // This does not print any values avoid accidentally leaking any sensitive env vars
@@ -168,6 +176,7 @@ impl From<InnerStore> for EnvVarsStore {
             create_gin_indexes: x.create_gin_indexes,
             use_brin_for_all_query_types: x.use_brin_for_all_query_types,
             disable_block_cache_for_lookup: x.disable_block_cache_for_lookup,
+            last_rollup_from_poi: x.last_rollup_from_poi,
         }
     }
 }
@@ -229,6 +238,8 @@ pub struct InnerStore {
     use_brin_for_all_query_types: bool,
     #[envconfig(from = "GRAPH_STORE_DISABLE_BLOCK_CACHE_FOR_LOOKUP", default = "false")]
     disable_block_cache_for_lookup: bool,
+    #[envconfig(from = "GRAPH_STORE_LAST_ROLLUP_FROM_POI", default = "false")]
+    last_rollup_from_poi: bool,
 }
 
 #[derive(Clone, Copy, Debug)]

store/postgres/src/deployment_store.rs

@@ -910,7 +910,11 @@ impl DeploymentStore {
 
         let mut conn = self.get_conn()?;
         let layout = store.layout(&mut conn, site.cheap_clone())?;
-        layout.block_time(&mut conn, block)
+        if ENV_VARS.store.last_rollup_from_poi {
+            layout.block_time(&mut conn, block)
+        } else {
+            layout.last_rollup(&mut conn)
+        }
     }
 
     pub(crate) async fn supports_proof_of_indexing<'a>(

store/postgres/src/relational.rs

@@ -1039,6 +1039,20 @@ impl Layout {
         Ok(block_time)
     }
 
+    /// Find the time of the last rollup for the subgraph. We do this by
+    /// looking for the maximum timestamp in any aggregation table and
+    /// adding a little bit more than the corresponding interval to it. This
+    /// method crucially depends on the fact that we always write the rollup
+    /// for all aggregations, meaning that if some aggregations do not have
+    /// an entry with the maximum timestamp that there was just no data for
+    /// that interval, but we did try to aggregate at that time.
+    pub(crate) fn last_rollup(
+        &self,
+        conn: &mut PgConnection,
+    ) -> Result<Option<BlockTime>, StoreError> {
+        Rollup::last_rollup(&self.rollups, conn)
+    }
+
     /// Construct `Rolllup` for each of the aggregation mappings
     /// `schema.agg_mappings()` and return them in the same order as the
     /// aggregation mappings

store/postgres/src/relational/rollup.rs

@@ -60,7 +60,7 @@ use std::sync::Arc;
 
 use diesel::{sql_query, PgConnection, RunQueryDsl as _};
 
-use diesel::sql_types::{Integer, Timestamptz};
+use diesel::sql_types::{Integer, Nullable, Timestamptz};
 use graph::blockchain::BlockTime;
 use graph::components::store::{BlockNumber, StoreError};
 use graph::constraint_violation;
@@ -70,6 +70,7 @@ use graph::schema::{
 };
 use graph::sqlparser::ast as p;
 use graph::sqlparser::parser::ParserError;
+use itertools::Itertools;
 
 use crate::relational::Table;
 
@@ -229,6 +230,10 @@ pub(crate) struct Rollup {
     #[allow(dead_code)]
     agg_table: Arc<Table>,
     insert_sql: String,
+    /// A query that determines the last time a rollup was done. The query
+    /// finds the latest timestamp in the aggregation table and adds the
+    /// length of the aggregation interval to deduce the last rollup time
+    last_rollup_sql: String,
 }
 
 impl Rollup {
@@ -256,10 +261,12 @@ impl Rollup {
         );
         let mut insert_sql = String::new();
         sql.insert(&mut insert_sql)?;
+        let last_rollup_sql = sql.last_rollup();
         Ok(Self {
             interval,
             agg_table,
             insert_sql,
+            last_rollup_sql,
         })
     }
 
@@ -275,6 +282,32 @@ impl Rollup {
             .bind::<Integer, _>(block);
         query.execute(conn)
     }
+
+    pub(crate) fn last_rollup(
+        rollups: &[Rollup],
+        conn: &mut PgConnection,
+    ) -> Result<Option<BlockTime>, StoreError> {
+        #[derive(QueryableByName)]
+        #[diesel(check_for_backend(diesel::pg::Pg))]
+        struct BlockTimeRes {
+            #[diesel(sql_type = Nullable<Timestamptz>)]
+            last_rollup: Option<BlockTime>,
+        }
+
+        if rollups.is_empty() {
+            return Ok(None);
+        }
+
+        let union_all = rollups
+            .iter()
+            .map(|rollup| &rollup.last_rollup_sql)
+            .join(" union all ");
+        let query = format!("select max(last_rollup) as last_rollup from ({union_all}) as a");
+        let last_rollup = sql_query(&query)
+            .get_result::<BlockTimeRes>(conn)
+            .map(|res| res.last_rollup)?;
+        Ok(last_rollup)
+    }
 }
 
 struct RollupSql<'a> {
@@ -479,6 +512,19 @@ impl<'a> RollupSql<'a> {
             self.insert_bucket(w)
         }
     }
+
+    /// Generate a query that selects the timestamp of the last rollup
+    fn last_rollup(&self) -> String {
+        // The timestamp column contains the timestamp of the start of the
+        // last bucket. The last rollup was therefore at least
+        // `self.interval` after that. We add 1 second to make sure we are
+        // well within the next bucket
+        let secs = self.interval.as_duration().as_secs() + 1;
+        format!(
+            "select max(timestamp) + '{} s'::interval as last_rollup from {}",
+            secs, self.agg_table.qualified_name
+        )
+    }
 }
 
 /// Write the elements in `list` separated by commas into `w`. The list