perf(server): rate-limit the active event stream count scan (#21951)

## 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.
This commit is contained in:
Charles Bochet
2026-06-22 16:58:32 +02:00
committed by GitHub
parent 5ce91e711c
commit 4dd9253d01
@@ -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({