graphql: Wrap responses in an Arc

Author: leoyvens

graph/Cargo.toml

@@ -33,7 +33,7 @@ num-bigint = { version = "^0.2.6", features = ["serde"] }
 num-traits = "0.2"
 rand = "0.6.1"
 semver = "0.10.0"
-serde = "1.0"
+serde = { version = "1.0", features = ["rc"] }
 serde_derive = "1.0"
 serde_json = { version = "1.0", features = ["arbitrary_precision"] }
 serde_yaml = "0.8"

graph/src/components/graphql.rs

@@ -12,7 +12,7 @@ use graphql_parser::query as q;
 use std::sync::Arc;
 
 /// Future for query results.
-pub type QueryResultFuture = Box<dyn Future<Item = QueryResult, Error = QueryError> + Send>;
+pub type QueryResultFuture = Box<dyn Future<Item = Arc<QueryResult>, Error = QueryError> + Send>;
 
 /// Future for subscription results.
 pub type SubscriptionResultFuture =
@@ -42,6 +42,8 @@ pub trait GraphQlRunner: Send + Sync + 'static {
             .await
             .map_err(move |e| format_err!("Failed to query metadata: {}", e))
             .and_then(move |result| {
+                // Metadata queries are not cached.
+                let result = Arc::try_unwrap(result).unwrap();
                 if result.errors.is_some() {
                     Err(format_err!("Failed to query metadata: {:?}", result.errors))
                 } else {

graph/src/data/query/error.rs

@@ -263,19 +263,23 @@ impl From<StoreError> for QueryExecutionError {
 #[derive(Debug)]
 pub enum QueryError {
     EncodingError(FromUtf8Error),
-    ParseError(q::ParseError),
+    ParseError(failure::Error),
     ExecutionError(QueryExecutionError),
 }
 
-impl From<FromUtf8Error> for QueryError {
-    fn from(e: FromUtf8Error) -> Self {
-        QueryError::EncodingError(e)
+impl Clone for QueryError {
+    fn clone(&self) -> Self {
+        match self {
+            QueryError::EncodingError(e) => QueryError::EncodingError(e.clone()),
+            QueryError::ParseError(e) => QueryError::ParseError(failure::err_msg(e.to_string())),
+            QueryError::ExecutionError(e) => QueryError::ExecutionError(e.clone()),
+        }
     }
 }
 
-impl From<q::ParseError> for QueryError {
-    fn from(e: q::ParseError) -> Self {
-        QueryError::ParseError(e)
+impl From<FromUtf8Error> for QueryError {
+    fn from(e: FromUtf8Error) -> Self {
+        QueryError::EncodingError(e)
     }
 }
 

graph/src/data/query/result.rs

@@ -13,7 +13,7 @@ where
 }
 
 /// The result of running a query, if successful.
-#[derive(Debug, Serialize)]
+#[derive(Debug, Clone, Serialize)]
 pub struct QueryResult {
     #[serde(
         skip_serializing_if = "Option::is_none",
@@ -30,17 +30,54 @@ pub struct QueryResult {
 }
 
 impl QueryResult {
+    /// A result with an empty object as the data.
+    pub fn empty() -> Self {
+        QueryResult {
+            data: Some(q::Value::Object(BTreeMap::new())),
+            errors: None,
+            extensions: None,
+        }
+    }
+
     pub fn new(data: Option<q::Value>) -> Self {
         QueryResult {
             data,
             errors: None,
             extensions: None,
         }
     }
+
     pub fn with_extensions(mut self, extensions: BTreeMap<q::Name, q::Value>) -> Self {
         self.extensions = Some(q::Value::Object(extensions));
         self
     }
+
+    pub fn has_errors(&self) -> bool {
+        return self.errors.is_some();
+    }
+
+    pub fn append(&mut self, mut other: QueryResult) {
+        match (&mut self.data, &mut other.data) {
+            (Some(q::Value::Object(ours)), Some(q::Value::Object(other))) => ours.append(other),
+
+            // Subgraph queries always return objects.
+            (Some(_), Some(_)) => unreachable!(),
+
+            // Only one side has data, use that.
+            _ => self.data = self.data.take().or(other.data),
+        }
+
+        match (&mut self.errors, &mut other.errors) {
+            (Some(ours), Some(other)) => ours.append(other),
+
+            // Only one side has errors, use that.
+            _ => self.errors = self.errors.take().or(other.errors),
+        }
+
+        // Currently we don't used extensions, the desired behaviour for merging them is tbd.
+        assert!(self.extensions.is_none());
+        assert!(other.extensions.is_none());
+    }
 }
 
 impl From<QueryExecutionError> for QueryResult {
@@ -61,20 +98,17 @@ impl From<Vec<QueryExecutionError>> for QueryResult {
     }
 }
 
-impl From<Result<q::Value, Vec<QueryExecutionError>>> for QueryResult {
-    fn from(result: Result<q::Value, Vec<QueryExecutionError>>) -> Self {
-        match result {
-            Ok(v) => QueryResult::new(Some(v)),
-            Err(errors) => QueryResult::from(errors),
-        }
+impl From<BTreeMap<String, q::Value>> for QueryResult {
+    fn from(val: BTreeMap<String, q::Value>) -> Self {
+        QueryResult::new(Some(q::Value::Object(val)))
     }
 }
 
-impl From<Result<BTreeMap<String, q::Value>, Vec<QueryExecutionError>>> for QueryResult {
-    fn from(result: Result<BTreeMap<String, q::Value>, Vec<QueryExecutionError>>) -> Self {
+impl<V: Into<QueryResult>, E: Into<QueryResult>> From<Result<V, E>> for QueryResult {
+    fn from(result: Result<V, E>) -> Self {
         match result {
-            Ok(v) => QueryResult::new(Some(q::Value::Object(v))),
-            Err(errors) => QueryResult::from(errors),
+            Ok(v) => v.into(),
+            Err(e) => e.into(),
         }
     }
 }

graph/src/data/subscription/result.rs

@@ -1,8 +1,10 @@
 use crate::prelude::QueryResult;
 use std::marker::Unpin;
+use std::sync::Arc;
 
 /// A stream of query results for a subscription.
-pub type QueryResultStream = Box<dyn futures03::stream::Stream<Item = QueryResult> + Send + Unpin>;
+pub type QueryResultStream =
+    Box<dyn futures03::stream::Stream<Item = Arc<QueryResult>> + Send + Unpin>;
 
 /// The result of running a subscription, if successful.
 pub type SubscriptionResult = QueryResultStream;

graphql/src/execution/cache.rs

@@ -110,7 +110,7 @@ pub struct QueryCache<R> {
     cache: Arc<Mutex<HashMap<Hash, Arc<CacheEntryInner<R>>>>>,
 }
 
-impl<R: Clone> QueryCache<R> {
+impl<R: CheapClone> QueryCache<R> {
     pub fn new() -> Self {
         Self {
             cache: Arc::new(Mutex::new(HashMap::new())),
@@ -131,7 +131,7 @@ impl<R: Clone> QueryCache<R> {
                     // Another thread is doing the work, release the lock and wait for it.
                     let entry = entry.get().cheap_clone();
                     drop(cache);
-                    return entry.wait().clone();
+                    return entry.wait().cheap_clone();
                 }
                 Entry::Vacant(entry) => {
                     let uncached = CacheEntryInner::new(hash);
@@ -153,7 +153,7 @@ impl<R: Clone> QueryCache<R> {
 
         // Actually compute the value and then share it with waiters.
         let value = f();
-        work.set(value.clone());
+        work.set(value.cheap_clone());
         value
     }
 }

graphql/src/execution/execution.rs

@@ -25,14 +25,14 @@ use crate::values::coercion;
 
 type QueryHash = <SetHasher as StableHasher>::Out;
 
-type QueryResponse = Result<BTreeMap<String, q::Value>, Vec<QueryExecutionError>>;
+pub(crate) type QueryResponse = Result<BTreeMap<String, q::Value>, Vec<QueryExecutionError>>;
 
 #[derive(Debug)]
 struct CacheByBlock {
     block: EthereumBlockPointer,
     max_weight: usize,
     weight: usize,
-    cache: HashMap<QueryHash, QueryResponse>,
+    cache: HashMap<QueryHash, Arc<QueryResult>>,
 }
 
 impl CacheByBlock {
@@ -46,9 +46,10 @@ impl CacheByBlock {
     }
 
     /// Returns `true` if the insert was successful or `false` if the cache was full.
-    fn insert(&mut self, key: QueryHash, value: QueryResponse) -> bool {
-        // Unwrap: We never try to insert errors into this cache.
-        let weight = value.as_ref().unwrap().weight();
+    fn insert(&mut self, key: QueryHash, value: Arc<QueryResult>) -> bool {
+        // We never try to insert errors into this cache, and always resolve some value.
+        assert!(value.errors.is_none());
+        let weight = value.data.as_ref().unwrap().weight();
         let fits_in_cache = self.weight + weight <= self.max_weight;
         if fits_in_cache {
             self.weight += weight;
@@ -95,7 +96,7 @@ lazy_static! {
 
     // This `VecDeque` works as a ring buffer with a capacity of `QUERY_CACHE_BLOCKS`.
     static ref QUERY_CACHE: RwLock<VecDeque<CacheByBlock>> = RwLock::new(VecDeque::new());
-    static ref QUERY_HERD_CACHE: QueryCache<QueryResponse> = QueryCache::new();
+    static ref QUERY_HERD_CACHE: QueryCache<Arc<QueryResult>> = QueryCache::new();
 }
 
 struct HashableQuery<'a> {
@@ -181,7 +182,7 @@ where
     pub query: Arc<crate::execution::Query>,
 
     /// The resolver to use.
-    pub resolver: Arc<R>,
+    pub resolver: R,
 
     /// Time at which the query times out.
     pub deadline: Option<Instant>,
@@ -230,7 +231,7 @@ where
 
         ExecutionContext {
             logger: self.logger.cheap_clone(),
-            resolver: Arc::new(introspection_resolver),
+            resolver: introspection_resolver,
             query: self.query.as_introspection_query(),
             deadline: self.deadline,
             max_first: std::u32::MAX,
@@ -295,17 +296,17 @@ pub fn execute_root_selection_set_uncached(
 }
 
 /// Executes the root selection set of a query.
-pub fn execute_root_selection_set(
-    ctx: &ExecutionContext<impl Resolver>,
+pub fn execute_root_selection_set<R: Resolver>(
+    ctx: &ExecutionContext<R>,
     selection_set: &q::SelectionSet,
     root_type: &s::ObjectType,
     block_ptr: Option<EthereumBlockPointer>,
-) -> QueryResponse {
+) -> Arc<QueryResult> {
     // Cache the cache key to not have to calculate it twice - once for lookup
     // and once for insert.
     let mut key: Option<QueryHash> = None;
 
-    if *CACHE_ALL || CACHED_SUBGRAPH_IDS.contains(&ctx.query.schema.id) {
+    if R::CACHEABLE && (*CACHE_ALL || CACHED_SUBGRAPH_IDS.contains(&ctx.query.schema.id)) {
         if let Some(block_ptr) = block_ptr {
             // JSONB and metadata queries use `BLOCK_NUMBER_MAX`. Ignore this case for two reasons:
             // - Metadata queries are not cacheable.
@@ -331,22 +332,30 @@ pub fn execute_root_selection_set(
 
     let result = if let Some(key) = key {
         QUERY_HERD_CACHE.cached_query(key, || {
-            execute_root_selection_set_uncached(ctx, selection_set, root_type)
+            Arc::new(QueryResult::from(execute_root_selection_set_uncached(
+                ctx,
+                selection_set,
+                root_type,
+            )))
         })
     } else {
-        execute_root_selection_set_uncached(ctx, selection_set, root_type)
+        Arc::new(QueryResult::from(execute_root_selection_set_uncached(
+            ctx,
+            selection_set,
+            root_type,
+        )))
     };
 
     // Check if this query should be cached.
     // Share errors from the herd cache, but don't store them in generational cache.
     // In particular, there is a problem where asking for a block pointer beyond the chain
     // head can cause the legitimate cache to be thrown out.
-    if let (Ok(_), Some(key), Some(block_ptr)) = (&result, key, block_ptr) {
+    if let (false, Some(key), Some(block_ptr)) = (result.has_errors(), key, block_ptr) {
         let mut cache = QUERY_CACHE.write().unwrap();
 
         // If there is already a cache by the block of this query, just add it there.
         if let Some(cache_by_block) = cache.iter_mut().find(|c| c.block == block_ptr) {
-            let cache_insert = cache_by_block.insert(key, result.clone());
+            let cache_insert = cache_by_block.insert(key, result.cheap_clone());
             ctx.cache_insert.store(cache_insert, Ordering::SeqCst);
         } else if *QUERY_CACHE_BLOCKS > 0 {
             // We're creating a new `CacheByBlock` if:
@@ -368,7 +377,7 @@ pub fn execute_root_selection_set(
                 // Create a new cache by block, insert this entry, and add it to the QUERY_CACHE.
                 let max_weight = *QUERY_CACHE_MAX_MEM / *QUERY_CACHE_BLOCKS;
                 let mut cache_by_block = CacheByBlock::new(block_ptr, max_weight);
-                let cache_insert = cache_by_block.insert(key, result.clone());
+                let cache_insert = cache_by_block.insert(key, result.cheap_clone());
                 ctx.cache_insert.store(cache_insert, Ordering::SeqCst);
                 cache.push_front(cache_by_block);
             }

graphql/src/execution/query.rs

@@ -122,9 +122,8 @@ impl Query {
         Ok(Arc::new(query))
     }
 
-    /// Return the block constraint for the toplevel query field(s) Since,
-    /// syntactically, each toplevel field can have its own block constraint,
-    /// we check that they are all identical and report an error otherwise
+    /// Return the block constraint for the toplevel query field(s), merging the
+    /// selection sets of fields that have the same block constraint.
     pub fn block_constraint(
         &self,
     ) -> Result<HashMap<BlockConstraint, q::SelectionSet>, Vec<QueryExecutionError>> {

graphql/src/execution/resolver.rs

@@ -61,7 +61,9 @@ impl<'a> ObjectOrInterface<'a> {
 }
 
 /// A GraphQL resolver that can resolve entities, enum values, scalar types and interfaces/unions.
-pub trait Resolver: Clone + Send + Sync {
+pub trait Resolver: Sized + Send + Sync {
+    const CACHEABLE: bool;
+
     /// Prepare for executing a query by prefetching as much data as possible
     fn prefetch(
         &self,

graphql/src/introspection/resolver.rs

@@ -348,6 +348,10 @@ impl<'a> IntrospectionResolver<'a> {
 
 /// A GraphQL resolver that can resolve entities, enum values, scalar types and interfaces/unions.
 impl<'a> Resolver for IntrospectionResolver<'a> {
+    // `IntrospectionResolver` is not used as a "top level" resolver,
+    // see `fn as_introspection_context`, so this value is irrelevant.
+    const CACHEABLE: bool = false;
+
     fn prefetch(
         &self,
         _: &ExecutionContext<Self>,