Split the block cache into block pointer cache and block data cache (#6037)

* split block cache

* get timestamp from pointers table

* update justfile

* use blocktime and fix tests

* code review updates

* use psql information_schema to check for migration requirements

Author: mangas

chain/ethereum/src/chain.rs

@@ -400,8 +400,9 @@ impl Chain {
     pub async fn block_number(
         &self,
         hash: &BlockHash,
-    ) -> Result<Option<(String, BlockNumber, Option<u64>, Option<BlockHash>)>, StoreError> {
-        self.chain_store.block_number(hash).await
+    ) -> Result<Option<(String, BlockNumber, Option<BlockTime>, Option<BlockHash>)>, StoreError>
+    {
+        self.chain_store.block_pointer(hash).await
     }
 
     // TODO: This is only used to build the block stream which could prolly
@@ -1130,6 +1131,9 @@ pub struct FirehoseMapper {
 impl BlockStreamMapper<Chain> for FirehoseMapper {
     fn decode_block(
         &self,
+        // We share the trait with substreams but for firehose the timestamp
+        // is in the block header so we don't need to use it here.
+        _timestamp: BlockTime,
         output: Option<&[u8]>,
     ) -> Result<Option<BlockFinality>, BlockStreamError> {
         let block = match output {
@@ -1198,12 +1202,19 @@ impl FirehoseMapperTrait<Chain> for FirehoseMapper {
         // Check about adding basic information about the block in the firehose::Response or maybe
         // define a slimmed down stuct that would decode only a few fields and ignore all the rest.
         let block = codec::Block::decode(any_block.value.as_ref())?;
+        let timestamp = block
+            .header()
+            .timestamp
+            .map(|ts| BlockTime::since_epoch(ts.seconds, ts.nanos as u32))
+            .unwrap_or_default();
 
         use firehose::ForkStep::*;
         match step {
             StepNew => {
                 // unwrap: Input cannot be None so output will be error or block.
-                let block = self.decode_block(Some(any_block.value.as_ref()))?.unwrap();
+                let block = self
+                    .decode_block(timestamp, Some(any_block.value.as_ref()))?
+                    .unwrap();
                 let block_with_triggers = self.block_with_triggers(logger, block).await?;
 
                 Ok(BlockStreamEvent::ProcessBlock(

chain/near/src/chain.rs

@@ -3,8 +3,8 @@ use graph::blockchain::client::ChainClient;
 use graph::blockchain::firehose_block_ingestor::FirehoseBlockIngestor;
 use graph::blockchain::substreams_block_stream::SubstreamsBlockStream;
 use graph::blockchain::{
-    BasicBlockchainBuilder, BlockIngestor, BlockchainBuilder, BlockchainKind, NoopDecoderHook,
-    NoopRuntimeAdapter, Trigger, TriggerFilterWrapper,
+    BasicBlockchainBuilder, BlockIngestor, BlockTime, BlockchainBuilder, BlockchainKind,
+    NoopDecoderHook, NoopRuntimeAdapter, Trigger, TriggerFilterWrapper,
 };
 use graph::cheap_clone::CheapClone;
 use graph::components::network_provider::ChainName;
@@ -432,6 +432,7 @@ pub struct FirehoseMapper {
 impl BlockStreamMapper<Chain> for FirehoseMapper {
     fn decode_block(
         &self,
+        _timestamp: BlockTime,
         output: Option<&[u8]>,
     ) -> Result<Option<codec::Block>, BlockStreamError> {
         let block = match output {
@@ -528,7 +529,10 @@ impl FirehoseMapperTrait<Chain> for FirehoseMapper {
         // Check about adding basic information about the block in the bstream::BlockResponseV2 or maybe
         // define a slimmed down stuct that would decode only a few fields and ignore all the rest.
         // unwrap: Input cannot be None so output will be error or block.
-        let block = self.decode_block(Some(any_block.value.as_ref()))?.unwrap();
+        let block = self
+            // the block time is inside the block.
+            .decode_block(BlockTime::MIN, Some(any_block.value.as_ref()))?
+            .unwrap();
 
         use ForkStep::*;
         match step {

chain/substreams/src/chain.rs

@@ -43,6 +43,7 @@ pub enum ParsedChanges {
 pub struct Block {
     pub hash: BlockHash,
     pub number: BlockNumber,
+    pub timestamp: BlockTime,
     pub changes: EntityChanges,
     pub parsed_changes: Vec<ParsedChanges>,
 }
@@ -60,7 +61,7 @@ impl blockchain::Block for Block {
     }
 
     fn timestamp(&self) -> BlockTime {
-        BlockTime::NONE
+        self.timestamp
     }
 }
 

chain/substreams/src/mapper.rs

@@ -32,6 +32,7 @@ pub struct WasmBlockMapper {
 impl BlockStreamMapper<Chain> for WasmBlockMapper {
     fn decode_block(
         &self,
+        _timestamp: BlockTime,
         _output: Option<&[u8]>,
     ) -> Result<Option<crate::Block>, BlockStreamError> {
         unreachable!("WasmBlockMapper does not do block decoding")
@@ -104,7 +105,11 @@ pub struct Mapper {
 
 #[async_trait]
 impl BlockStreamMapper<Chain> for Mapper {
-    fn decode_block(&self, output: Option<&[u8]>) -> Result<Option<Block>, BlockStreamError> {
+    fn decode_block(
+        &self,
+        timestamp: BlockTime,
+        output: Option<&[u8]>,
+    ) -> Result<Option<Block>, BlockStreamError> {
         let changes: EntityChanges = match output {
             Some(msg) => Message::decode(msg).map_err(SubstreamsError::DecodingError)?,
             None => EntityChanges {
@@ -125,6 +130,7 @@ impl BlockStreamMapper<Chain> for Mapper {
             number,
             changes,
             parsed_changes,
+            timestamp,
         };
 
         Ok(Some(block))
@@ -152,9 +158,13 @@ impl BlockStreamMapper<Chain> for Mapper {
     ) -> Result<BlockStreamEvent<Chain>, BlockStreamError> {
         let block_number: BlockNumber = clock.number.try_into().map_err(Error::from)?;
         let block_hash = clock.id.as_bytes().to_vec().into();
+        let timestamp = clock
+            .timestamp
+            .map(|ts| BlockTime::since_epoch(ts.seconds, ts.nanos as u32))
+            .unwrap_or_default();
 
         let block = self
-            .decode_block(Some(&block))?
+            .decode_block(timestamp, Some(&block))?
             .ok_or_else(|| anyhow!("expected block to not be empty"))?;
 
         let block = self.block_with_triggers(logger, block).await.map(|bt| {

graph/src/blockchain/block_stream.rs

@@ -685,7 +685,11 @@ pub trait FirehoseMapper<C: Blockchain>: Send + Sync {
 
 #[async_trait]
 pub trait BlockStreamMapper<C: Blockchain>: Send + Sync {
-    fn decode_block(&self, output: Option<&[u8]>) -> Result<Option<C::Block>, BlockStreamError>;
+    fn decode_block(
+        &self,
+        timestamp: BlockTime,
+        output: Option<&[u8]>,
+    ) -> Result<Option<C::Block>, BlockStreamError>;
 
     async fn block_with_triggers(
         &self,

graph/src/blockchain/mock.rs

@@ -546,10 +546,11 @@ impl ChainStore for MockChainStore {
     fn confirm_block_hash(&self, _number: BlockNumber, _hash: &BlockHash) -> Result<usize, Error> {
         unimplemented!()
     }
-    async fn block_number(
+    async fn block_pointer(
         &self,
         _hash: &BlockHash,
-    ) -> Result<Option<(String, BlockNumber, Option<u64>, Option<BlockHash>)>, StoreError> {
+    ) -> Result<Option<(String, BlockNumber, Option<BlockTime>, Option<BlockHash>)>, StoreError>
+    {
         unimplemented!()
     }
     async fn block_numbers(

graph/src/blockchain/types.rs

@@ -1,4 +1,5 @@
 use anyhow::anyhow;
+use chrono::DateTime;
 use diesel::deserialize::FromSql;
 use diesel::pg::Pg;
 use diesel::serialize::{Output, ToSql};
@@ -7,6 +8,7 @@ use diesel::sql_types::{Bytea, Nullable, Text};
 use diesel_derives::{AsExpression, FromSqlRow};
 use serde::{Deserialize, Deserializer};
 use std::convert::TryFrom;
+use std::num::ParseIntError;
 use std::time::Duration;
 use std::{fmt, str::FromStr};
 use web3::types::{Block, H256, U256, U64};
@@ -16,9 +18,9 @@ use crate::components::store::BlockNumber;
 use crate::data::graphql::IntoValue;
 use crate::data::store::scalar::Timestamp;
 use crate::derive::CheapClone;
-use crate::object;
 use crate::prelude::{r, Value};
 use crate::util::stable_hash_glue::{impl_stable_hash, AsBytes};
+use crate::{bail, object};
 
 /// A simple marker for byte arrays that are really block hashes
 #[derive(Clone, Default, PartialEq, Eq, Hash, FromSqlRow, AsExpression)]
@@ -477,10 +479,7 @@ impl TryFrom<(Option<H256>, Option<U64>, H256, U256)> for ExtendedBlockPtr {
         let block_number =
             i32::try_from(number).map_err(|_| anyhow!("Block number out of range"))?;
 
-        // Convert `U256` to `BlockTime`
-        let secs =
-            i64::try_from(timestamp_u256).map_err(|_| anyhow!("Timestamp out of range for i64"))?;
-        let block_time = BlockTime::since_epoch(secs, 0);
+        let block_time = BlockTime::try_from(timestamp_u256)?;
 
         Ok(ExtendedBlockPtr {
             hash: hash.into(),
@@ -497,16 +496,13 @@ impl TryFrom<(H256, i32, H256, U256)> for ExtendedBlockPtr {
     fn try_from(tuple: (H256, i32, H256, U256)) -> Result<Self, Self::Error> {
         let (hash, block_number, parent_hash, timestamp_u256) = tuple;
 
-        // Convert `U256` to `BlockTime`
-        let secs =
-            i64::try_from(timestamp_u256).map_err(|_| anyhow!("Timestamp out of range for i64"))?;
-        let block_time = BlockTime::since_epoch(secs, 0);
+        let timestamp = BlockTime::try_from(timestamp_u256)?;
 
         Ok(ExtendedBlockPtr {
             hash: hash.into(),
             number: block_number,
             parent_hash: parent_hash.into(),
-            timestamp: block_time,
+            timestamp,
         })
     }
 }
@@ -562,14 +558,63 @@ impl fmt::Display for ChainIdentifier {
 #[diesel(sql_type = Timestamptz)]
 pub struct BlockTime(Timestamp);
 
+impl Default for BlockTime {
+    fn default() -> Self {
+        BlockTime::NONE
+    }
+}
+
+impl TryFrom<BlockTime> for U256 {
+    type Error = anyhow::Error;
+
+    fn try_from(value: BlockTime) -> Result<Self, Self::Error> {
+        if value.as_secs_since_epoch() < 0 {
+            bail!("unable to convert block time into U256");
+        }
+
+        Ok(U256::from(value.as_secs_since_epoch() as u64))
+    }
+}
+
+impl TryFrom<U256> for BlockTime {
+    type Error = anyhow::Error;
+
+    fn try_from(value: U256) -> Result<Self, Self::Error> {
+        i64::try_from(value)
+            .map_err(|_| anyhow!("Timestamp out of range for i64"))
+            .map(|ts| BlockTime::since_epoch(ts, 0))
+    }
+}
+
+impl TryFrom<Option<String>> for BlockTime {
+    type Error = ParseIntError;
+
+    fn try_from(ts: Option<String>) -> Result<Self, Self::Error> {
+        match ts {
+            Some(str) => return BlockTime::from_hex_str(&str),
+            None => return Ok(BlockTime::NONE),
+        };
+    }
+}
+
 impl BlockTime {
     /// A timestamp from a long long time ago used to indicate that we don't
     /// have a timestamp
-    pub const NONE: Self = Self(Timestamp::NONE);
+    pub const NONE: Self = Self::MIN;
 
     pub const MAX: Self = Self(Timestamp::MAX);
 
-    pub const MIN: Self = Self(Timestamp::MIN);
+    pub const MIN: Self = Self(Timestamp(DateTime::from_timestamp_nanos(0)));
+
+    pub fn from_hex_str(ts: &str) -> Result<Self, ParseIntError> {
+        let (radix, idx) = if ts.starts_with("0x") {
+            (16, 2)
+        } else {
+            (10, 0)
+        };
+
+        u64::from_str_radix(&ts[idx..], radix).map(|ts| BlockTime::since_epoch(ts as i64, 0))
+    }
 
     /// Construct a block time that is the given number of seconds and
     /// nanoseconds after the Unix epoch
@@ -586,7 +631,12 @@ impl BlockTime {
     /// hourly rollups in tests
     #[cfg(debug_assertions)]
     pub fn for_test(ptr: &BlockPtr) -> Self {
-        Self::since_epoch(ptr.number as i64 * 45 * 60, 0)
+        Self::for_test_number(&ptr.number)
+    }
+
+    #[cfg(debug_assertions)]
+    pub fn for_test_number(number: &BlockNumber) -> Self {
+        Self::since_epoch(*number as i64 * 45 * 60, 0)
     }
 
     pub fn as_secs_since_epoch(&self) -> i64 {

graph/src/components/store/traits.rs

@@ -576,10 +576,10 @@ pub trait ChainStore: ChainHeadStore {
     /// Currently, the timestamp is only returned if it's present in the top level block. This format is
     /// depends on the chain and the implementation of Blockchain::Block for the specific chain.
     /// eg: {"block": { "timestamp": 123123123 } }
-    async fn block_number(
+    async fn block_pointer(
         &self,
         hash: &BlockHash,
-    ) -> Result<Option<(String, BlockNumber, Option<u64>, Option<BlockHash>)>, StoreError>;
+    ) -> Result<Option<(String, BlockNumber, Option<BlockTime>, Option<BlockHash>)>, StoreError>;
 
     /// Do the same lookup as `block_number`, but in bulk
     async fn block_numbers(
@@ -665,10 +665,10 @@ pub trait QueryStore: Send + Sync {
     /// Returns the blocknumber, timestamp and the parentHash. Timestamp depends on the chain block type
     /// and can have multiple formats, it can also not be prevent. For now this is only available
     /// for EVM chains both firehose and rpc.
-    async fn block_number_with_timestamp_and_parent_hash(
+    async fn block_pointer(
         &self,
         block_hash: &BlockHash,
-    ) -> Result<Option<(BlockNumber, Option<u64>, Option<BlockHash>)>, StoreError>;
+    ) -> Result<Option<(BlockNumber, Option<BlockTime>, Option<BlockHash>)>, StoreError>;
 
     fn wait_stats(&self) -> PoolWaitStats;
 

graphql/src/runner.rs

@@ -111,11 +111,11 @@ where
         let latest_block = match store.block_ptr().await.ok().flatten() {
             Some(block) => Some(LatestBlockInfo {
                 timestamp: store
-                    .block_number_with_timestamp_and_parent_hash(&block.hash)
+                    .block_pointer(&block.hash)
                     .await
                     .ok()
                     .flatten()
-                    .and_then(|(_, t, _)| t),
+                    .and_then(|(_, t, _)| t.map(|ts| ts.as_secs_since_epoch() as u64)),
                 hash: block.hash,
                 number: block.number,
             }),

graphql/src/store/resolver.rs

@@ -186,7 +186,7 @@ impl StoreResolver {
         let (timestamp, parent_hash) = if lookup_needed(field) {
             match self
                 .store
-                .block_number_with_timestamp_and_parent_hash(&block_ptr.hash)
+                .block_pointer(&block_ptr.hash)
                 .await
                 .map_err(Into::<QueryExecutionError>::into)?
             {
@@ -219,7 +219,7 @@ impl StoreResolver {
             .unwrap_or(r::Value::Null);
 
         let timestamp = timestamp
-            .map(|ts| r::Value::Int(ts as i64))
+            .map(|ts| r::Value::Int(ts.as_secs_since_epoch()))
             .unwrap_or(r::Value::Null);
 
         let parent_hash = parent_hash

justfile

@@ -0,0 +1,40 @@
+DATABASE_TEST_VAR_NAME := "THEGRAPH_STORE_POSTGRES_DIESEL_URL"
+DATABASE_URL := "postgresql://graph-node:let-me-in@localhost:5432/graph-node"
+
+
+help:
+    @just -l
+
+local-deps-up *ARGS:
+    docker compose -f docker/docker-compose.yml up ipfs postgres {{ ARGS }}
+
+local-deps-down:
+    docker compose -f docker/docker-compose.yml down
+
+test-deps-up *ARGS:
+    docker compose -f tests/docker-compose.yml up {{ ARGS }}
+
+test-deps-down:
+    docker compose -f tests/docker-compose.yml down
+
+# Requires local-deps, see local-deps-up
+test *ARGS:
+    just _run_in_bash cargo test --workspace --exclude graph-tests -- --nocapture {{ ARGS }}
+
+runner-test *ARGS:
+    just _run_in_bash cargo test -p graph-tests --test runner_tests -- --nocapture {{ ARGS }}
+
+# Requires test-deps to be running, see test-deps-up
+it-test *ARGS:
+    just _run_in_bash cargo test --test integration_tests -- --nocapture {{ ARGS }}
+
+local-rm-db:
+    rm -r docker/data/postgres
+
+new-migration NAME:
+    diesel migration generate {{ NAME }} --migration-dir store/postgres/migrations/
+
+_run_in_bash *CMD:
+    #!/usr/bin/env bash
+    export {{ DATABASE_TEST_VAR_NAME }}={{ DATABASE_URL }}
+    {{ CMD }}