fix: Reset max queue length and reducer wait time metrics

Max value metrics are currently reset to zero after each collection period.
However this won't be reflected in the metrics store immediately.

Once the database records new values for said metrics,
only then will the reset be reflected in the metrics store.

This subtlety is imperceptible for metrics that are frequently updated.
But if a metric is not frequently updated,
or for periods of reduced load in general,
the metrics store may present stale max value metrics.

This anomaly was thought to have been fixed by #709.
However it only fixed it for DB_METRICS.
This patch removes the anomaly for WORKER_METRICS as well.

Author: joshua-spacetime

crates/core/src/host/scheduler.rs

@@ -1,12 +1,10 @@
-use std::collections::hash_map::DefaultHasher;
-use std::hash::Hasher;
 use std::path::Path;
 
 use futures::StreamExt;
 use rustc_hash::FxHashMap;
 use sled::transaction::{ConflictableTransactionError::Abort as TxAbort, TransactionError};
+use spacetimedb_lib::bsatn;
 use spacetimedb_lib::bsatn::ser::BsatnError;
-use spacetimedb_lib::{bsatn, Address};
 use tokio::sync::mpsc;
 use tokio_util::time::delay_queue::Expired;
 use tokio_util::time::{delay_queue, DelayQueue};
@@ -273,18 +271,11 @@ impl SchedulerActor {
         };
         let scheduled: ScheduledReducer = bsatn::from_slice(&scheduled).unwrap();
 
-        fn hash(a: &Address, b: &str) -> u64 {
-            use std::hash::Hash;
-            let mut hasher = DefaultHasher::new();
-            a.hash(&mut hasher);
-            b.hash(&mut hasher);
-            hasher.finish()
-        }
-
         let db = module_host.info().address;
+        let reducer = scheduled.reducer.clone();
         let mut guard = MAX_REDUCER_DELAY.lock().unwrap();
         let max_reducer_delay = *guard
-            .entry(hash(&db, &scheduled.reducer))
+            .entry((db, reducer))
             .and_modify(|max| {
                 if delay > *max {
                     *max = delay;

crates/core/src/worker_metrics/mod.rs

@@ -84,11 +84,17 @@ metrics_group!(
     }
 );
 
+type ReducerLabel = (Address, String);
+
 pub static MAX_QUEUE_LEN: Lazy<Mutex<HashMap<Address, i64>>> = Lazy::new(|| Mutex::new(HashMap::new()));
-pub static MAX_REDUCER_DELAY: Lazy<Mutex<HashMap<u64, f64>>> = Lazy::new(|| Mutex::new(HashMap::new()));
+pub static MAX_REDUCER_DELAY: Lazy<Mutex<HashMap<ReducerLabel, f64>>> = Lazy::new(|| Mutex::new(HashMap::new()));
 pub static WORKER_METRICS: Lazy<WorkerMetrics> = Lazy::new(WorkerMetrics::new);
 
 pub fn reset_counters() {
+    // Reset max queue length
+    WORKER_METRICS.instance_queue_length_max.0.reset();
     MAX_QUEUE_LEN.lock().unwrap().clear();
+    // Reset max reducer wait time
+    WORKER_METRICS.scheduled_reducer_delay_sec_max.0.reset();
     MAX_REDUCER_DELAY.lock().unwrap().clear();
 }