Aggregations (write side)

chain/arweave/src/codec.rs

@@ -2,7 +2,11 @@
 #[path = "protobuf/sf.arweave.r#type.v1.rs"]
 mod pbcodec;
 
-use graph::{blockchain::Block as BlockchainBlock, blockchain::BlockPtr, prelude::BlockNumber};
+use graph::{
+    blockchain::Block as BlockchainBlock,
+    blockchain::{BlockPtr, BlockTime},
+    prelude::BlockNumber,
+};
 
 pub use pbcodec::*;
 
@@ -28,6 +32,10 @@ impl BlockchainBlock for Block {
             number: self.number().saturating_sub(1),
         })
     }
+
+    fn timestamp(&self) -> BlockTime {
+        BlockTime::since_epoch(i64::try_from(self.timestamp).unwrap(), 0)
+    }
 }
 
 impl AsRef<[u8]> for BigInt {

chain/arweave/src/data_source.rs

@@ -94,6 +94,7 @@ impl blockchain::DataSource<Chain> for DataSource {
             trigger.cheap_clone(),
             handler.clone(),
             block.ptr(),
+            block.timestamp(),
         )))
     }
 

chain/cosmos/src/codec.rs

@@ -1,6 +1,6 @@
 pub(crate) use crate::protobuf::pbcodec::*;
 
-use graph::blockchain::Block as BlockchainBlock;
+use graph::blockchain::{Block as BlockchainBlock, BlockTime};
 use graph::{
     blockchain::BlockPtr,
     prelude::{anyhow::anyhow, BlockNumber, Error},
@@ -80,6 +80,11 @@ impl BlockchainBlock for Block {
     fn parent_ptr(&self) -> Option<BlockPtr> {
         self.parent_ptr().unwrap()
     }
+
+    fn timestamp(&self) -> BlockTime {
+        let time = self.header().unwrap().time.as_ref().unwrap();
+        BlockTime::since_epoch(time.seconds, time.nanos as u32)
+    }
 }
 
 impl HeaderOnlyBlock {
@@ -137,6 +142,11 @@ impl BlockchainBlock for HeaderOnlyBlock {
     fn parent_ptr(&self) -> Option<BlockPtr> {
         self.parent_ptr().unwrap()
     }
+
+    fn timestamp(&self) -> BlockTime {
+        let time = self.header().unwrap().time.as_ref().unwrap();
+        BlockTime::since_epoch(time.seconds, time.nanos as u32)
+    }
 }
 
 impl EventData {

chain/cosmos/src/data_source.rs

@@ -127,6 +127,7 @@ impl blockchain::DataSource<Chain> for DataSource {
             trigger.cheap_clone(),
             handler,
             block.ptr(),
+            block.timestamp(),
         )))
     }
 

chain/ethereum/src/chain.rs

@@ -2,7 +2,7 @@ use anyhow::{anyhow, bail, Result};
 use anyhow::{Context, Error};
 use graph::blockchain::client::ChainClient;
 use graph::blockchain::firehose_block_ingestor::{FirehoseBlockIngestor, Transforms};
-use graph::blockchain::{BlockIngestor, BlockchainKind, TriggersAdapterSelector};
+use graph::blockchain::{BlockIngestor, BlockTime, BlockchainKind, TriggersAdapterSelector};
 use graph::components::store::DeploymentCursorTracker;
 use graph::data::subgraph::UnifiedMappingApiVersion;
 use graph::firehose::{FirehoseEndpoint, ForkStep};
@@ -588,6 +588,15 @@ impl Block for BlockFinality {
             BlockFinality::NonFinal(block) => json::to_value(&block.ethereum_block),
         }
     }
+
+    fn timestamp(&self) -> BlockTime {
+        let ts = match self {
+            BlockFinality::Final(block) => block.timestamp,
+            BlockFinality::NonFinal(block) => block.ethereum_block.block.timestamp,
+        };
+        let ts = i64::try_from(ts.as_u64()).unwrap();
+        BlockTime::since_epoch(ts, 0)
+    }
 }
 
 pub struct DummyDataSourceTemplate;

chain/ethereum/src/codec.rs

@@ -4,7 +4,9 @@ mod pbcodec;
 
 use anyhow::format_err;
 use graph::{
-    blockchain::{Block as BlockchainBlock, BlockPtr, ChainStoreBlock, ChainStoreData},
+    blockchain::{
+        self, Block as BlockchainBlock, BlockPtr, BlockTime, ChainStoreBlock, ChainStoreData,
+    },
     prelude::{
         web3,
         web3::types::{Bytes, H160, H2048, H256, H64, U256, U64},
@@ -449,6 +451,11 @@ impl BlockchainBlock for Block {
     fn data(&self) -> Result<jsonrpc_core::serde_json::Value, jsonrpc_core::serde_json::Error> {
         self.header().to_json()
     }
+
+    fn timestamp(&self) -> BlockTime {
+        let ts = self.header().timestamp.as_ref().unwrap();
+        BlockTime::since_epoch(ts.seconds, ts.nanos as u32)
+    }
 }
 
 impl HeaderOnlyBlock {
@@ -502,6 +509,11 @@ impl BlockchainBlock for HeaderOnlyBlock {
     fn data(&self) -> Result<jsonrpc_core::serde_json::Value, jsonrpc_core::serde_json::Error> {
         self.header().to_json()
     }
+
+    fn timestamp(&self) -> blockchain::BlockTime {
+        let ts = self.header().timestamp.as_ref().unwrap();
+        blockchain::BlockTime::since_epoch(ts.seconds, ts.nanos as u32)
+    }
 }
 
 #[cfg(test)]

chain/ethereum/src/data_source.rs

@@ -606,11 +606,11 @@ impl DataSource {
     fn matches_trigger_address(&self, trigger: &EthereumTrigger) -> bool {
         let Some(ds_address) = self.address else {
             // 'wildcard' data sources match any trigger address.
-            return true
+            return true;
         };
 
         let Some(trigger_address) = trigger.address() else {
-             return true
+            return true;
         };
 
         ds_address == *trigger_address
@@ -644,6 +644,7 @@ impl DataSource {
                     },
                     handler.handler,
                     block.block_ptr(),
+                    block.timestamp(),
                 )))
             }
             EthereumTrigger::Log(log_ref) => {
@@ -753,6 +754,7 @@ impl DataSource {
                     },
                     event_handler.handler,
                     block.block_ptr(),
+                    block.timestamp(),
                     logging_extras,
                 )))
             }
@@ -863,6 +865,7 @@ impl DataSource {
                     },
                     handler.handler,
                     block.block_ptr(),
+                    block.timestamp(),
                     logging_extras,
                 )))
             }

chain/near/src/codec.rs

@@ -8,7 +8,7 @@ pub mod substreams_triggers;
 
 use graph::{
     blockchain::Block as BlockchainBlock,
-    blockchain::BlockPtr,
+    blockchain::{BlockPtr, BlockTime},
     prelude::{hex, web3::types::H256, BlockNumber},
 };
 use std::convert::TryFrom;
@@ -75,6 +75,11 @@ impl BlockchainBlock for Block {
     fn parent_ptr(&self) -> Option<BlockPtr> {
         self.parent_ptr()
     }
+
+    fn timestamp(&self) -> BlockTime {
+        let ts = i64::try_from(self.header().timestamp).unwrap();
+        BlockTime::since_epoch(ts, 0)
+    }
 }
 
 impl HeaderOnlyBlock {
@@ -101,6 +106,11 @@ impl BlockchainBlock for HeaderOnlyBlock {
     fn parent_ptr(&self) -> Option<BlockPtr> {
         self.header().parent_ptr()
     }
+
+    fn timestamp(&self) -> BlockTime {
+        let ts = i64::try_from(self.header().timestamp).unwrap();
+        BlockTime::since_epoch(ts, 0)
+    }
 }
 
 impl execution_outcome::Status {

chain/near/src/data_source.rs

@@ -163,6 +163,7 @@ impl blockchain::DataSource<Chain> for DataSource {
             trigger.cheap_clone(),
             handler.clone(),
             block.ptr(),
+            block.timestamp(),
         )))
     }
 

chain/starknet/src/codec.rs

@@ -28,4 +28,8 @@ impl BlockchainBlock for Block {
             })
         }
     }
+
+    fn timestamp(&self) -> graph::blockchain::BlockTime {
+        graph::blockchain::BlockTime::since_epoch(self.timestamp as i64, 0)
+    }
 }

chain/starknet/src/data_source.rs

@@ -148,6 +148,7 @@ impl blockchain::DataSource<Chain> for DataSource {
             trigger.clone(),
             handler,
             block.ptr(),
+            block.timestamp(),
         )))
     }
 

chain/substreams/examples/substreams.rs

@@ -88,7 +88,7 @@ async fn main() -> Result<(), Error> {
             Some(event) => match event {
                 Err(_) => {}
                 Ok(block_stream_event) => match block_stream_event {
-                    BlockStreamEvent::ProcessWasmBlock(_, _, _, _) => {
+                    BlockStreamEvent::ProcessWasmBlock(_, _, _, _, _) => {
                         unreachable!("Cannot happen with this mapper")
                     }
                     BlockStreamEvent::Revert(_, _) => {}

chain/substreams/src/block_ingestor.rs

@@ -74,7 +74,13 @@ impl SubstreamsBlockIngestor {
 
         while let Some(message) = stream.next().await {
             let (block, cursor) = match message {
-                Ok(BlockStreamEvent::ProcessWasmBlock(_block_ptr, _data, _handler, _cursor)) => {
+                Ok(BlockStreamEvent::ProcessWasmBlock(
+                    _block_ptr,
+                    _block_time,
+                    _data,
+                    _handler,
+                    _cursor,
+                )) => {
                     unreachable!("Block ingestor should never receive raw blocks");
                 }
                 Ok(BlockStreamEvent::ProcessBlock(triggers, cursor)) => {

chain/substreams/src/chain.rs

@@ -3,7 +3,7 @@ use crate::{data_source::*, EntityChanges, TriggerData, TriggerFilter, TriggersA
 use anyhow::Error;
 use graph::blockchain::client::ChainClient;
 use graph::blockchain::{
-    BasicBlockchainBuilder, BlockIngestor, EmptyNodeCapabilities, NoopRuntimeAdapter,
+    BasicBlockchainBuilder, BlockIngestor, BlockTime, EmptyNodeCapabilities, NoopRuntimeAdapter,
 };
 use graph::components::store::DeploymentCursorTracker;
 use graph::env::EnvVars;
@@ -57,6 +57,10 @@ impl blockchain::Block for Block {
     fn parent_ptr(&self) -> Option<BlockPtr> {
         None
     }
+
+    fn timestamp(&self) -> BlockTime {
+        BlockTime::NONE
+    }
 }
 
 pub struct Chain {

chain/substreams/src/mapper.rs

@@ -6,13 +6,15 @@ use anyhow::{anyhow, Error};
 use graph::blockchain::block_stream::{
     BlockStreamEvent, BlockStreamMapper, BlockWithTriggers, FirehoseCursor, SubstreamsError,
 };
+use graph::blockchain::BlockTime;
 use graph::data::store::scalar::Bytes;
 use graph::data::store::IdType;
 use graph::data::value::Word;
 use graph::data_source::CausalityRegion;
 use graph::prelude::{async_trait, BigInt, BlockHash, BlockNumber, Logger, Value};
 use graph::prelude::{BigDecimal, BlockPtr};
 use graph::schema::InputSchema;
+use graph::slog::error;
 use graph::substreams::Clock;
 use prost::Message;
 
@@ -41,15 +43,15 @@ impl BlockStreamMapper<Chain> for WasmBlockMapper {
 
     async fn handle_substreams_block(
         &self,
-        _logger: &Logger,
+        logger: &Logger,
         clock: Clock,
         cursor: FirehoseCursor,
         block: Vec<u8>,
     ) -> Result<BlockStreamEvent<Chain>, Error> {
         let Clock {
             id,
             number,
-            timestamp: _,
+            timestamp,
         } = clock;
 
         let block_ptr = BlockPtr {
@@ -59,8 +61,24 @@ impl BlockStreamMapper<Chain> for WasmBlockMapper {
 
         let block_data = block.into_boxed_slice();
 
+        // `timestamp` is an `Option`, but it should always be set
+        let timestamp = match timestamp {
+            None => {
+                error!(logger,
+                    "Substream block is missing a timestamp";
+                    "cursor" => cursor.to_string(),
+                    "number" => number,
+                );
+                return Err(anyhow!(
+                    "Substream block is missing a timestamp at cursor {cursor}, block number {number}"
+                ));
+            }
+            Some(ts) => BlockTime::since_epoch(ts.seconds, ts.nanos as u32),
+        };
+
         Ok(BlockStreamEvent::ProcessWasmBlock(
             block_ptr,
+            timestamp,
             block_data,
             self.handler.clone(),
             cursor,

core/src/subgraph/context/mod.rs

@@ -6,7 +6,7 @@ use crate::polling_monitor::{
 use anyhow::{self, Error};
 use bytes::Bytes;
 use graph::{
-    blockchain::Blockchain,
+    blockchain::{BlockTime, Blockchain},
     components::{
         store::{DeploymentId, SubgraphFork},
         subgraph::{HostMetrics, MappingError, RuntimeHost as _, SharedProofOfIndexing},
@@ -133,6 +133,7 @@ impl<C: Blockchain, T: RuntimeHostBuilder<C>> IndexingContext<C, T> {
         &self,
         logger: &Logger,
         block_ptr: BlockPtr,
+        block_time: BlockTime,
         block_data: Box<[u8]>,
         handler: String,
         mut state: BlockState<C>,
@@ -161,6 +162,7 @@ impl<C: Blockchain, T: RuntimeHostBuilder<C>> IndexingContext<C, T> {
             .process_block(
                 logger,
                 block_ptr,
+                block_time,
                 block_data,
                 handler,
                 state,