graph, runtime: Override user-supplied timestamp in store.set

For timeseries, users can not set the timestamp. Instead, the time for the
current block is used.

Author: lutter

graph/src/blockchain/types.rs

@@ -350,6 +350,12 @@ impl BlockTime {
     }
 }
 
+impl From<BlockTime> for Value {
+    fn from(block_time: BlockTime) -> Self {
+        Value::Int8(block_time.as_secs_since_epoch())
+    }
+}
+
 impl TryFrom<&Value> for BlockTime {
     type Error = anyhow::Error;
 

graph/src/schema/input_schema.rs

@@ -262,7 +262,7 @@ pub struct ObjectType {
     /// The name of the aggregation to which this object type belongs if it
     /// is part of an aggregation
     aggregation: Option<Atom>,
-    timeseries: bool,
+    pub timeseries: bool,
     interfaces: Box<[Word]>,
     shared_interfaces: Box<[Atom]>,
 }

runtime/test/src/test.rs

@@ -1,3 +1,4 @@
+use graph::blockchain::BlockTime;
 use graph::components::metrics::gas::GasMetrics;
 use graph::data::store::{scalar, Id, IdType};
 use graph::data::subgraph::*;
@@ -1608,7 +1609,7 @@ async fn test_store_ts() {
         timestamp: Int8!
         amount: BigDecimal!
     }
-    
+
     type Stats @aggregation(intervals: ["hour"], source: "Data") {
         id: Bytes!
         timestamp: Int8!
@@ -1617,15 +1618,27 @@ async fn test_store_ts() {
 
     let mut host = Host::new(schema, "hostStoreTs", "boolean.wasm", None).await;
 
+    let block_time = host.ctx.timestamp;
+    let other_time = BlockTime::since_epoch(7000, 0);
+    // If this fails, something is wrong with the test setup
+    assert_ne!(block_time, other_time);
+
     let b20 = Value::BigDecimal(20.into());
 
     host.store_setv(
         DATA,
         DID,
-        vec![("timestamp", Value::Int8(23)), ("amount", b20.clone())],
+        vec![
+            ("timestamp", Value::from(other_time)),
+            ("amount", b20.clone()),
+        ],
     )
     .expect("Setting 'Data' is allowed");
 
+    // Set overrides the user-supplied timestamp for timeseries
+    let data = host.store_get(DATA, DID).unwrap().unwrap();
+    assert_eq!(Some(&Value::from(block_time)), data.get("timestamp"));
+
     let err = host
         .store_setv(STATS, SID, vec![("amount", b20)])
         .expect_err("store_set must fail for aggregations");

runtime/wasm/src/host_exports.rs

@@ -13,7 +13,7 @@ use semver::Version;
 use wasmtime::Trap;
 use web3::types::H160;
 
-use graph::blockchain::Blockchain;
+use graph::blockchain::{BlockTime, Blockchain};
 use graph::components::store::{EnsLookup, GetScope, LoadRelatedRequest};
 use graph::components::subgraph::{
     PoICausalityRegion, ProofOfIndexingEvent, SharedProofOfIndexing,
@@ -218,6 +218,7 @@ impl<C: Blockchain> HostExports<C> {
         state: &mut BlockState<C>,
         block: BlockNumber,
         proof_of_indexing: &SharedProofOfIndexing,
+        block_time: BlockTime,
         entity_type: String,
         entity_id: String,
         mut data: HashMap<Word, Value>,
@@ -251,6 +252,10 @@ impl<C: Blockchain> HostExports<C> {
             "store_set",
         )?;
 
+        if entity_type.object_type()?.timeseries {
+            data.insert(Word::from("timestamp"), block_time.into());
+        }
+
         // Set the id if there isn't one yet, and make sure that a
         // previously set id agrees with the one in the `key`
         match data.get(&store::ID) {
@@ -1095,7 +1100,7 @@ pub mod test_support {
     use std::{borrow::Cow, collections::HashMap, sync::Arc};
 
     use graph::{
-        blockchain::Blockchain,
+        blockchain::{BlockTime, Blockchain},
         components::{
             store::{BlockNumber, GetScope},
             subgraph::SharedProofOfIndexing,
@@ -1108,11 +1113,17 @@ pub mod test_support {
 
     use crate::MappingContext;
 
-    pub struct HostExports<C: Blockchain>(Arc<super::HostExports<C>>);
+    pub struct HostExports<C: Blockchain> {
+        host_exports: Arc<super::HostExports<C>>,
+        block_time: BlockTime,
+    }
 
     impl<C: Blockchain> HostExports<C> {
         pub fn new(ctx: &MappingContext<C>) -> Self {
-            HostExports(ctx.host_exports.clone())
+            HostExports {
+                host_exports: ctx.host_exports.clone(),
+                block_time: ctx.timestamp,
+            }
         }
 
         pub fn store_set(
@@ -1127,11 +1138,12 @@ pub mod test_support {
             stopwatch: &StopwatchMetrics,
             gas: &GasCounter,
         ) -> Result<(), HostExportError> {
-            self.0.store_set(
+            self.host_exports.store_set(
                 logger,
                 state,
                 block,
                 proof_of_indexing,
+                self.block_time,
                 entity_type,
                 entity_id,
                 data,
@@ -1147,7 +1159,7 @@ pub mod test_support {
             entity_id: String,
             gas: &GasCounter,
         ) -> Result<Option<Cow<'a, Entity>>, anyhow::Error> {
-            self.0
+            self.host_exports
                 .store_get(state, entity_type, entity_id, gas, GetScope::Store)
         }
     }

runtime/wasm/src/module/mod.rs

@@ -1154,6 +1154,7 @@ impl<C: Blockchain> WasmInstanceContext<C> {
             &mut self.ctx.state,
             self.ctx.block_ptr.number,
             &self.ctx.proof_of_indexing,
+            self.ctx.timestamp,
             entity,
             id,
             data,