From 6cc03a6c3be8e145525e3be477c14a67c1fe0a50 Mon Sep 17 00:00:00 2001
From: phil <uniphil@gmail.com>
Date: Thu, 12 Jun 2025 12:37:25 -0400
Subject: [PATCH] metrics + less logging for slow inserts

---
 ufos/src/storage.rs | 17 +++++++++++++----
 1 file changed, 13 insertions(+), 4 deletions(-)

diff --git a/ufos/src/storage.rs b/ufos/src/storage.rs
index a3b2bf4..222da1b 100644
--- a/ufos/src/storage.rs
+++ b/ufos/src/storage.rs
@@ -5,9 +5,10 @@ use crate::{
 };
 use async_trait::async_trait;
 use jetstream::exports::{Did, Nsid};
+use metrics::{describe_histogram, histogram, Unit};
 use std::collections::{HashMap, HashSet};
 use std::path::Path;
-use std::time::{Duration, SystemTime};
+use std::time::{Duration, Instant};
 use tokio::sync::mpsc::Receiver;
 use tokio_util::sync::CancellationToken;
 
@@ -35,21 +36,29 @@ where
         self,
         mut batches: Receiver<EventBatch<LIMIT>>,
     ) -> StorageResult<()> {
+        describe_histogram!(
+            "storage_slow_batches",
+            Unit::Microseconds,
+            "batches that took more than 3s to insert"
+        );
         while let Some(event_batch) = batches.recv().await {
             let token = CancellationToken::new();
             let cancelled = token.clone();
             tokio::spawn(async move {
-                let started = SystemTime::now();
+                let started = Instant::now();
                 let mut concerned = false;
                 loop {
                     tokio::select! {
-                        _ = tokio::time::sleep(Duration::from_secs_f64(3.)) => {
-                            log::warn!("taking a long time to insert an event batch ({:?})...", started.elapsed());
+                        _ = tokio::time::sleep(Duration::from_secs(3)) => {
+                            if !concerned {
+                                log::warn!("taking a long time to insert an event batch...");
+                            }
                             concerned = true;
                         }
                         _ = cancelled.cancelled() => {
                             if concerned {
                                 log::warn!("finally inserted slow event batch (or failed) after {:?}", started.elapsed());
+                                histogram!("storage_slow_batches").record(started.elapsed().as_micros() as f64);
                             }
                             break
                         }
-- 
2.51.2