all: Address review comments for load management

Comments are in Zac's review of
#1762

Author: lutter

graph/src/data/graphql/effort.rs

@@ -96,8 +96,7 @@ impl QueryEffort {
         let query_effort = inner
             .effort
             .get(&shape_hash)
-            .map(|stats| stats.average())
-            .flatten();
+            .and_then(|stats| stats.average());
         (query_effort, total_effort)
     }
 }

graph/src/util/stats.rs

@@ -65,7 +65,7 @@ impl Bin {
     fn average_gt(&self, duration: Duration) -> bool {
         // Compute self.duration / self.count > duration as
         // self.duration > duration * self.count. If the RHS
-        // oveflows, we assume the average would have been smaller
+        // overflows, we assume the average would have been smaller
         // than any duration
         duration
             .checked_mul(self.count)
@@ -99,12 +99,20 @@ impl Default for MovingStats {
 }
 
 impl MovingStats {
+    /// Track moving statistics over a window of `window_size` duration
+    /// and keep the measurements in bins of `bin_size` each.
+    ///
+    /// # Panics
+    ///
+    /// Panics if `window_size` or `bin_size` is `0`, or if `bin_size` >=
+    /// `window_size`
     pub fn new(window_size: Duration, bin_size: Duration) -> Self {
-        let capacity = if bin_size.as_millis() > 0 {
-            window_size.as_millis() as usize / bin_size.as_millis() as usize
-        } else {
-            1
-        };
+        assert!(window_size.as_millis() > 0);
+        assert!(bin_size.as_millis() > 0);
+        assert!(window_size > bin_size);
+
+        let capacity = window_size.as_millis() as usize / bin_size.as_millis() as usize;
+
         MovingStats {
             window_size,
             bin_size,

node/src/main.rs

@@ -528,7 +528,7 @@ async fn main() {
         store_conn_pool_size,
         &logger,
         connection_pool_registry,
-        wait_stats.clone(),
+        wait_stats.cheap_clone(),
     );
 
     let chain_head_update_listener = Arc::new(PostgresChainHeadUpdateListener::new(

store/postgres/src/connection_pool.rs

@@ -6,7 +6,7 @@ use graph::util::security::SafeDisplay;
 
 use std::collections::HashMap;
 use std::fmt;
-use std::sync::{Arc, RwLock};
+use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
 
 struct ErrorHandler(Logger, Box<Counter>);
@@ -29,7 +29,7 @@ struct EventHandler {
     count_gauge: Box<Gauge>,
     wait_gauge: Box<Gauge>,
     wait_stats: PoolWaitStats,
-    last_log: RwLock<Instant>,
+    last_log: Mutex<Instant>,
 }
 
 impl EventHandler {
@@ -53,14 +53,14 @@ impl EventHandler {
             count_gauge,
             wait_gauge,
             wait_stats,
-            last_log: RwLock::new(Instant::now()),
+            last_log: Mutex::new(Instant::now()),
         }
     }
 
     fn add_wait_time(&self, duration: Duration) {
         let should_log = {
             // Log average wait time, but at most every 10s
-            let mut last_log = self.last_log.write().unwrap();
+            let mut last_log = self.last_log.lock().unwrap();
             if last_log.elapsed() > Duration::from_secs(10) {
                 *last_log = Instant::now();
                 true
@@ -110,7 +110,7 @@ pub fn create_connection_pool(
     pool_size: u32,
     logger: &Logger,
     registry: Arc<dyn MetricsRegistry>,
-    wait_time: Arc<RwLock<MovingStats>>,
+    wait_time: PoolWaitStats,
 ) -> Pool<ConnectionManager<PgConnection>> {
     let logger_store = logger.new(o!("component" => "Store"));
     let logger_pool = logger.new(o!("component" => "PostgresConnectionPool"));

store/test-store/src/lib.rs

@@ -393,7 +393,6 @@ fn execute_subgraph_query_internal(
 ) -> QueryResult {
     let logger = Logger::root(slog::Discard, o!());
     let query = return_err!(PreparedQuery::new(query, max_complexity, 100));
-    //let effort =
     let mut values = std::collections::BTreeMap::new();
     let mut errors = Vec::new();
     for (bc, selection_set) in return_err!(query.block_constraint()) {