From 4dd9253d010557b400acf3d0336db2fcdcd8783c Mon Sep 17 00:00:00 2001
From: Charles Bochet <charlesBochet@users.noreply.github.com>
Date: Mon, 22 Jun 2026 16:58:32 +0200
Subject: [PATCH] perf(server): rate-limit the active event stream count scan
 (#21951)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

## Context

`twenty_event_streams_live_total` (an observable gauge) calls
`getTotalActiveStreamCount()` →
`scanAndCountSetMembers('workspace:*:activeStreams')`, which runs a
full-keyspace `SCAN MATCH` over the entire Redis DB **on every metrics
scrape, on every pod**. `SCAN MATCH` walks every key (filtering only the
output), and the subscriptions namespace shares the node with the
workspace metadata cache (~190k keys in our prod), so this was the
dominant Redis command (billions of `SCAN` calls) to count a handful of
sets.

## What this does

Cache the count and refresh it via the scan at most once per
`ACTIVE_STREAM_COUNT_REFRESH_MS` (5 min) per pod, instead of on every
scrape. Steady-state scrapes return the cached value; the authoritative
scan still runs periodically so the gauge stays fresh.

Single-method change; no new Redis keys, no data-model changes.
---
 .../subscriptions/event-stream.service.ts     | 20 ++++++++++++++++---
 1 file changed, 17 insertions(+), 3 deletions(-)

diff --git a/packages/twenty-server/src/engine/subscriptions/event-stream.service.ts b/packages/twenty-server/src/engine/subscriptions/event-stream.service.ts
index f3292c6481..5dbe44d9f0 100644
--- a/packages/twenty-server/src/engine/subscriptions/event-stream.service.ts
+++ b/packages/twenty-server/src/engine/subscriptions/event-stream.service.ts
@@ -19,9 +19,13 @@ import {
   type RecordOrMetadataGqlOperationSignature,
 } from 'src/engine/subscriptions/types/event-stream-data.type';
 
+const ACTIVE_STREAM_COUNT_REFRESH_MS = 5 * 60 * 1_000;
+
 @Injectable()
 export class EventStreamService implements OnModuleInit {
   private readonly logger = new Logger(EventStreamService.name);
+  private activeStreamCount = 0;
+  private activeStreamCountRefreshedAt = 0;
 
   constructor(
     @InjectCacheStorage(CacheStorageNamespace.EngineSubscriptions)
@@ -42,9 +46,19 @@ export class EventStreamService implements OnModuleInit {
   }
 
   async getTotalActiveStreamCount(): Promise<number> {
-    return this.cacheStorageService.scanAndCountSetMembers(
-      'workspace:*:activeStreams',
-    );
+    const now = Date.now();
+    const isStale =
+      now - this.activeStreamCountRefreshedAt >= ACTIVE_STREAM_COUNT_REFRESH_MS;
+
+    if (isStale) {
+      this.activeStreamCount =
+        await this.cacheStorageService.scanAndCountSetMembers(
+          'workspace:*:activeStreams',
+        );
+      this.activeStreamCountRefreshedAt = now;
+    }
+
+    return this.activeStreamCount;
   }
 
   async createEventStream({