From e2dde8268a770f8d185322efddd3a904de44e70e Mon Sep 17 00:00:00 2001
From: Quentin Gliech <quenting@element.io>
Date: Mon, 28 Apr 2025 10:45:10 +0200
Subject: [PATCH] Add a gauge which tracks the number of pending activity
 records

---
 crates/handlers/src/activity_tracker/worker.rs | 15 ++++++++++++++-
 1 file changed, 14 insertions(+), 1 deletion(-)

diff --git a/crates/handlers/src/activity_tracker/worker.rs b/crates/handlers/src/activity_tracker/worker.rs
index 427eb0b4c..80853f0fe 100644
--- a/crates/handlers/src/activity_tracker/worker.rs
+++ b/crates/handlers/src/activity_tracker/worker.rs
@@ -10,7 +10,7 @@ use chrono::{DateTime, Utc};
 use mas_storage::{RepositoryAccess, RepositoryError, user::BrowserSessionRepository};
 use opentelemetry::{
     Key, KeyValue,
-    metrics::{Counter, Histogram},
+    metrics::{Counter, Gauge, Histogram},
 };
 use sqlx::PgPool;
 use tokio_util::sync::CancellationToken;
@@ -45,6 +45,7 @@ struct ActivityRecord {
 pub struct Worker {
     pool: PgPool,
     pending_records: HashMap<(SessionKind, Ulid), ActivityRecord>,
+    pending_records_gauge: Gauge<u64>,
     message_counter: Counter<u64>,
     flush_time_histogram: Histogram<u64>,
 }
@@ -80,9 +81,17 @@ impl Worker {
             .with_unit("ms")
             .build();
 
+        let pending_records_gauge = METER
+            .u64_gauge("mas.activity_tracker.pending_records")
+            .with_description("The number of pending activity records")
+            .with_unit("{records}")
+            .build();
+        pending_records_gauge.record(0, &[]);
+
         Self {
             pool,
             pending_records: HashMap::with_capacity(MAX_PENDING_RECORDS),
+            pending_records_gauge,
             message_counter,
             flush_time_histogram,
         }
@@ -165,6 +174,10 @@ impl Worker {
                     let _ = tx.send(());
                 }
             }
+
+            // Update the gauge
+            self.pending_records_gauge
+                .record(self.pending_records.len() as u64, &[]);
         }
 
         // Flush one last time