Merge branch 'master' into fix-compliance-issues-with-gql-server

Author: YassinEldeeb

Cargo.lock

@@ -21,7 +21,7 @@ repository = "https://github.com/graphprotocol/graph-node"
 license = "MIT OR Apache-2.0"
 
 [workspace.dependencies]
-prost = "0.11.8"
+prost = "0.11.9"
 prost-types = "0.11.8"
 tonic = { version = "0.8.3", features = ["tls-roots", "gzip"] }
 tonic-build = { version = "0.8.4", features = ["prost"] }

Cargo.toml

@@ -26,7 +26,7 @@ graph-runtime-derive = { path = "../../runtime/derive" }
 
 [dev-dependencies]
 base64 = "0.20.0"
-uuid = { version = "1.3.0", features = ["v4"] }
+uuid = { version = "1.3.1", features = ["v4"] }
 
 [build-dependencies]
 tonic-build = { workspace = true }

chain/ethereum/Cargo.toml

@@ -138,13 +138,14 @@ impl FromAscObj<AscUnresolvedContractCall_0_0_4> for UnresolvedContractCall {
         asc_call: AscUnresolvedContractCall_0_0_4,
         heap: &H,
         gas: &GasCounter,
+        depth: usize,
     ) -> Result<Self, DeterministicHostError> {
         Ok(UnresolvedContractCall {
-            contract_name: asc_get(heap, asc_call.contract_name, gas)?,
-            contract_address: asc_get(heap, asc_call.contract_address, gas)?,
-            function_name: asc_get(heap, asc_call.function_name, gas)?,
-            function_signature: Some(asc_get(heap, asc_call.function_signature, gas)?),
-            function_args: asc_get(heap, asc_call.function_args, gas)?,
+            contract_name: asc_get(heap, asc_call.contract_name, gas, depth)?,
+            contract_address: asc_get(heap, asc_call.contract_address, gas, depth)?,
+            function_name: asc_get(heap, asc_call.function_name, gas, depth)?,
+            function_signature: Some(asc_get(heap, asc_call.function_signature, gas, depth)?),
+            function_args: asc_get(heap, asc_call.function_args, gas, depth)?,
         })
     }
 }
@@ -163,13 +164,14 @@ impl FromAscObj<AscUnresolvedContractCall> for UnresolvedContractCall {
         asc_call: AscUnresolvedContractCall,
         heap: &H,
         gas: &GasCounter,
+        depth: usize,
     ) -> Result<Self, DeterministicHostError> {
         Ok(UnresolvedContractCall {
-            contract_name: asc_get(heap, asc_call.contract_name, gas)?,
-            contract_address: asc_get(heap, asc_call.contract_address, gas)?,
-            function_name: asc_get(heap, asc_call.function_name, gas)?,
+            contract_name: asc_get(heap, asc_call.contract_name, gas, depth)?,
+            contract_address: asc_get(heap, asc_call.contract_address, gas, depth)?,
+            function_name: asc_get(heap, asc_call.function_name, gas, depth)?,
             function_signature: None,
-            function_args: asc_get(heap, asc_call.function_args, gas)?,
+            function_args: asc_get(heap, asc_call.function_args, gas, depth)?,
         })
     }
 }

chain/ethereum/src/runtime/abi.rs

@@ -77,9 +77,9 @@ fn ethereum_call(
     // function signature; subgraphs using an apiVersion < 0.0.4 don't pass
     // the signature along with the call.
     let call: UnresolvedContractCall = if ctx.heap.api_version() >= Version::new(0, 0, 4) {
-        asc_get::<_, AscUnresolvedContractCall_0_0_4, _>(ctx.heap, wasm_ptr.into(), &ctx.gas)?
+        asc_get::<_, AscUnresolvedContractCall_0_0_4, _>(ctx.heap, wasm_ptr.into(), &ctx.gas, 0)?
     } else {
-        asc_get::<_, AscUnresolvedContractCall, _>(ctx.heap, wasm_ptr.into(), &ctx.gas)?
+        asc_get::<_, AscUnresolvedContractCall, _>(ctx.heap, wasm_ptr.into(), &ctx.gas, 0)?
     };
 
     let result = eth_call(

chain/ethereum/src/runtime/runtime_adapter.rs

@@ -7,10 +7,10 @@ use graph::{
         store::{DeploymentLocator, EntityKey, EntityType, SubgraphFork},
         subgraph::{MappingError, ProofOfIndexingEvent, SharedProofOfIndexing},
     },
-    data::store::scalar::Bytes,
+    data::{store::scalar::Bytes, value::Word},
     data_source::{self, CausalityRegion},
     prelude::{
-        anyhow, async_trait, BigDecimal, BigInt, BlockHash, BlockNumber, BlockState, Entity,
+        anyhow, async_trait, BigDecimal, BigInt, BlockHash, BlockNumber, BlockState,
         RuntimeHostBuilder, Value,
     },
     slog::Logger,
@@ -189,7 +189,7 @@ where
                         entity_id: entity_id.clone().into(),
                         causality_region: CausalityRegion::ONCHAIN, // Substreams don't currently support offchain data
                     };
-                    let mut data: HashMap<String, Value> = HashMap::from_iter(vec![]);
+                    let mut data: HashMap<Word, Value> = HashMap::from_iter(vec![]);
 
                     for field in entity_change.fields.iter() {
                         let new_value: &codec::value::Typed = match &field.new_value {
@@ -200,21 +200,25 @@ where
                         };
 
                         let value: Value = decode_value(new_value)?;
-                        *data.entry(field.name.clone()).or_insert(Value::Null) = value;
+                        *data
+                            .entry(Word::from(field.name.clone()))
+                            .or_insert(Value::Null) = value;
                     }
 
                     write_poi_event(
                         proof_of_indexing,
                         &ProofOfIndexingEvent::SetEntity {
                             entity_type,
                             id: &entity_id,
+                            // TODO: This should be an entity so we do not have to build the intermediate HashMap
                             data: &data,
                         },
                         causality_region,
                         logger,
                     );
 
-                    state.entity_cache.set(key, Entity::from(data))?;
+                    let entity = state.entity_cache.make_entity(data)?;
+                    state.entity_cache.set(key, entity)?;
                 }
                 Operation::Delete => {
                     let entity_type: &str = &entity_change.entity;

chain/substreams/src/trigger.rs

@@ -36,4 +36,4 @@ ipfs-api-backend-hyper = "0.6"
 ipfs-api = { version = "0.17.0", features = [
     "with-hyper-rustls",
 ], default-features = false }
-uuid = { version = "1.3.0", features = ["v4"] }
+uuid = { version = "1.3.1", features = ["v4"] }

core/Cargo.toml

@@ -17,7 +17,7 @@ pub fn ipfs_service(
     client: IpfsClient,
     max_file_size: u64,
     timeout: Duration,
-    concurrency_and_rate_limit: u16,
+    rate_limit: u16,
 ) -> IpfsService {
     let ipfs = IpfsServiceInner {
         client,
@@ -26,8 +26,7 @@ pub fn ipfs_service(
     };
 
     let svc = ServiceBuilder::new()
-        .rate_limit(concurrency_and_rate_limit.into(), Duration::from_secs(1))
-        .concurrency_limit(concurrency_and_rate_limit as usize)
+        .rate_limit(rate_limit.into(), Duration::from_secs(1))
         .service_fn(move |req| ipfs.cheap_clone().call_inner(req))
         .boxed();
 

core/src/polling_monitor/ipfs_service.rs

@@ -161,6 +161,8 @@ where
 
                     // Object not found, push the id to the back of the queue.
                     Ok((id, None)) => {
+                        debug!(logger, "not found on polling"; "object_id" => id.to_string());
+
                         metrics.not_found.inc();
                         queue.push_back(id);
                     }

core/src/polling_monitor/mod.rs

@@ -33,3 +33,42 @@ pub struct IndexingInputs<C: Blockchain> {
     /// possibly expensive and noisy, information
     pub instrument: bool,
 }
+
+impl<C: Blockchain> IndexingInputs<C> {
+    pub fn with_store(&self, store: Arc<dyn WritableStore>) -> Self {
+        let IndexingInputs {
+            deployment,
+            features,
+            start_blocks,
+            stop_block,
+            store: _,
+            debug_fork,
+            triggers_adapter,
+            chain,
+            templates,
+            unified_api_version,
+            static_filters,
+            poi_version,
+            network,
+            manifest_idx_and_name,
+            instrument,
+        } = self;
+        IndexingInputs {
+            deployment: deployment.clone(),
+            features: features.clone(),
+            start_blocks: start_blocks.clone(),
+            stop_block: stop_block.clone(),
+            store,
+            debug_fork: debug_fork.clone(),
+            triggers_adapter: triggers_adapter.clone(),
+            chain: chain.clone(),
+            templates: templates.clone(),
+            unified_api_version: unified_api_version.clone(),
+            static_filters: *static_filters,
+            poi_version: *poi_version,
+            network: network.clone(),
+            manifest_idx_and_name: manifest_idx_and_name.clone(),
+            instrument: *instrument,
+        }
+    }
+}