houseassassin/twenty
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Replace Redis key scanning with sorted-set event stream tracking (#23326)

Browse Source 
## Context

The `twenty_event_streams_live_total` gauge counted live streams by
SCANning every `workspace:*:activeStreams` key and summing set
cardinalities. A full metric refresh walks the entire Redis keyspace, on
every server instance, and its cost grows with unrelated cache data
rather than with the number of streams.

## What changed

Adds a metric-only sorted set, `activeStreamExpirations`. Members are
`workspaceId:eventStreamChannelId`, scores are expiration timestamps:

- Create and successful heartbeat refresh: `ZADD` with score `now +
EVENT_STREAM_TTL_MS`
- Destroy and stale cleanup: `ZREM`
- Gauge read: `ZREMRANGEBYSCORE` + `ZCARD` in one transaction

The scan-and-count cache helper is removed.

Scores are written at the same moments the stream key TTL is set, so a
member expires exactly when its stream key would. Any missed cleanup
(crashed pod, failed heartbeat) resolves itself at the next gauge read.
Metric writes are best effort: failures are logged and never affect
stream creation, refresh, or cleanup. Existing stream keys and
application behavior are unchanged, and no migration is needed.

## Tradeoffs

- One extra `ZADD` per 30-second heartbeat
- During a rolling deploy, streams owned by old pods appear in the gauge
after their next heartbeat (undercount bounded by one heartbeat
interval)
- A destroy racing a concurrent refresh can leave one orphaned member
until its score lapses (gauge over-counts by 1 for at most one TTL)

## Testing

- Unit coverage for the sorted-set cache helpers and the stream
lifecycle (create, refresh success/failure, destroy, stale cleanup)
- `npx nx typecheck twenty-server`, targeted Oxlint, 14 tests passing
This commit is contained in:
Weiko
2026-07-27 18:35:54 +02:00
committed by GitHub
parent a66aacfb82
commit 24ccdb9b5d
4 changed files with 388 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/twenty-server/src/engine/core-modules/cache-storage/services/__tests__/cache-storage.service.spec.ts
+86
View File
@@ -109,4 +109,90 @@ describe('CacheStorageService', () => {
);
});
});
describe('sorted sets', () => {
const createRedisCacheMock = () => {
const zAdd = jest.fn().mockResolvedValue(1);
const zRem = jest.fn().mockResolvedValue(1);
const exec = jest.fn().mockResolvedValue([2, 3]);
const multi = {
zRemRangeByScore: jest.fn(),
zCard: jest.fn(),
exec,
};
multi.zRemRangeByScore.mockReturnValue(multi);
multi.zCard.mockReturnValue(multi);
const cache = {
store: {
name: 'redis',
client: {
zAdd,
zRem,
multi: jest.fn().mockReturnValue(multi),
},
},
} as unknown as Cache;
return { cache, exec, multi, zAdd, zRem };
};
it('adds members with scores', async () => {
const { cache, zAdd } = createRedisCacheMock();
const cacheStorageService = new CacheStorageService(
cache,
CacheStorageNamespace.EngineWorkspace,
);
const entries = [
{ score: 1000, value: 'stream-1' },
{ score: 2000, value: 'stream-2' },
];
await cacheStorageService.sortedSetAdd('active-streams', entries);
expect(zAdd).toHaveBeenCalledWith(prefixKey('active-streams'), entries);
});
it('removes members', async () => {
const { cache, zRem } = createRedisCacheMock();
const cacheStorageService = new CacheStorageService(
cache,
CacheStorageNamespace.EngineWorkspace,
);
await cacheStorageService.sortedSetRemove('active-streams', [
'stream-1',
'stream-2',
]);
expect(zRem).toHaveBeenCalledWith(prefixKey('active-streams'), [
'stream-1',
'stream-2',
]);
});
it('removes expired members and returns the remaining count atomically', async () => {
const { cache, exec, multi } = createRedisCacheMock();
const cacheStorageService = new CacheStorageService(
cache,
CacheStorageNamespace.EngineWorkspace,
);
const count = await cacheStorageService.sortedSetRemoveByScoreAndCount(
'active-streams',
0,
1000,
);
expect(multi.zRemRangeByScore).toHaveBeenCalledWith(
prefixKey('active-streams'),
0,
1000,
);
expect(multi.zCard).toHaveBeenCalledWith(prefixKey('active-streams'));
expect(exec).toHaveBeenCalledTimes(1);
expect(count).toBe(3);
});
});
});
packages/twenty-server/src/engine/core-modules/cache-storage/services/cache-storage.service.ts
+43 -32
View File
@@ -236,45 +236,56 @@ export class CacheStorageService {
} while (cursor !== 0);
}
async scanAndCountSetMembers(scanPattern: string): Promise<number> {
async sortedSetAdd(
key: string,
entries: Array<{ score: number; value: string }>,
): Promise<number> {
if (entries.length === 0) {
return 0;
}
if (!this.isRedisCache(this.cache)) {
throw new Error('sortedSetAdd is only supported with Redis cache');
}
return this.cache.store.client.zAdd(this.getKey(key), entries);
}
async sortedSetRemove(key: string, values: string[]): Promise<number> {
if (values.length === 0) {
return 0;
}
if (!this.isRedisCache(this.cache)) {
throw new Error('sortedSetRemove is only supported with Redis cache');
}
return this.cache.store.client.zRem(this.getKey(key), values);
}
async sortedSetRemoveByScoreAndCount(
key: string,
minScore: number,
maxScore: number,
): Promise<number> {
if (!this.isRedisCache(this.cache)) {
throw new Error(
'scanAndCountSetMembers is only supported with Redis cache',
'sortedSetRemoveByScoreAndCount is only supported with Redis cache',
);
}
const redisClient = this.cache.store.client;
let cursor = 0;
let totalCount = 0;
const prefixedKey = this.getKey(key);
const [, count] = await this.cache.store.client
.multi()
.zRemRangeByScore(prefixedKey, minScore, maxScore)
.zCard(prefixedKey)
.exec();
do {
const result = await redisClient.scan(cursor, {
MATCH: `${this.namespace}:${scanPattern}`,
COUNT: 100,
});
if (count instanceof Error) {
throw count;
}
cursor = result.cursor;
const keys = result.keys;
if (keys.length > 0) {
const pipeline = redisClient.multi();
for (const key of keys) {
pipeline.sCard(key);
}
const results = await pipeline.exec();
for (const result of results) {
if (result instanceof Error) {
throw result;
}
totalCount += result as number;
}
}
} while (cursor !== 0);
return totalCount;
return count as number;
}
async acquireLock(key: string, ttl = 1000): Promise<boolean> {
packages/twenty-server/src/engine/subscriptions/__tests__/event-stream.service.spec.ts
+187
View File
@@ -0,0 +1,187 @@
import { type CacheStorageService } from 'src/engine/core-modules/cache-storage/services/cache-storage.service';
import { type MetricsService } from 'src/engine/core-modules/metrics/metrics.service';
import { EVENT_STREAM_TTL_MS } from 'src/engine/subscriptions/constants/event-stream-ttl.constant';
import { EventStreamService } from 'src/engine/subscriptions/event-stream.service';
const WORKSPACE_ID = 'workspace-id';
const EVENT_STREAM_CHANNEL_ID = 'event-stream-channel-id';
const EVENT_STREAM_KEY = `eventStream:${WORKSPACE_ID}:${EVENT_STREAM_CHANNEL_ID}`;
const ACTIVE_STREAMS_KEY = `workspace:${WORKSPACE_ID}:activeStreams`;
const ACTIVE_STREAM_EXPIRATIONS_KEY = 'activeStreamExpirations';
const ACTIVE_STREAM_EXPIRATION_MEMBER = `${WORKSPACE_ID}:${EVENT_STREAM_CHANNEL_ID}`;
describe('EventStreamService', () => {
let cacheStorageService: jest.Mocked<CacheStorageService>;
let service: EventStreamService;
beforeEach(() => {
jest.useFakeTimers();
jest.setSystemTime(new Date('2026-07-26T12:00:00.000Z'));
cacheStorageService = {
del: jest.fn(),
expire: jest.fn(),
get: jest.fn(),
set: jest.fn(),
setAdd: jest.fn(),
setRemove: jest.fn(),
sortedSetAdd: jest.fn(),
sortedSetRemove: jest.fn(),
sortedSetRemoveByScoreAndCount: jest.fn(),
} as unknown as jest.Mocked<CacheStorageService>;
service = new EventStreamService(cacheStorageService, {
createObservableGauge: jest.fn(),
} as unknown as MetricsService);
});
afterEach(() => {
jest.useRealTimers();
});
it('counts live streams without scanning Redis keys', async () => {
cacheStorageService.sortedSetRemoveByScoreAndCount.mockResolvedValue(3);
await expect(service.getTotalActiveStreamCount()).resolves.toBe(3);
await expect(service.getTotalActiveStreamCount()).resolves.toBe(3);
expect(
cacheStorageService.sortedSetRemoveByScoreAndCount,
).toHaveBeenCalledTimes(1);
expect(
cacheStorageService.sortedSetRemoveByScoreAndCount,
).toHaveBeenCalledWith(ACTIVE_STREAM_EXPIRATIONS_KEY, 0, Date.now());
});
it('adds a stream to its functional and metric indexes on creation', async () => {
cacheStorageService.get.mockResolvedValue(undefined);
await service.createEventStream({
workspaceId: WORKSPACE_ID,
eventStreamChannelId: EVENT_STREAM_CHANNEL_ID,
authContext: { userId: 'user-id' },
});
expect(cacheStorageService.set).toHaveBeenCalledWith(
EVENT_STREAM_KEY,
{
authContext: { userId: 'user-id' },
workspaceId: WORKSPACE_ID,
queries: {},
createdAt: Date.now(),
},
EVENT_STREAM_TTL_MS,
);
expect(cacheStorageService.setAdd).toHaveBeenCalledWith(
ACTIVE_STREAMS_KEY,
[EVENT_STREAM_CHANNEL_ID],
EVENT_STREAM_TTL_MS,
);
expect(cacheStorageService.sortedSetAdd).toHaveBeenCalledWith(
ACTIVE_STREAM_EXPIRATIONS_KEY,
[
{
score: Date.now() + EVENT_STREAM_TTL_MS,
value: ACTIVE_STREAM_EXPIRATION_MEMBER,
},
],
);
});
it('tracks a stream after a successful refresh', async () => {
cacheStorageService.expire
.mockResolvedValueOnce(true)
.mockResolvedValueOnce(true);
await expect(
service.refreshEventStreamTTL({
workspaceId: WORKSPACE_ID,
eventStreamChannelId: EVENT_STREAM_CHANNEL_ID,
}),
).resolves.toBe(true);
expect(cacheStorageService.sortedSetAdd).toHaveBeenCalledWith(
ACTIVE_STREAM_EXPIRATIONS_KEY,
[
{
score: Date.now() + EVENT_STREAM_TTL_MS,
value: ACTIVE_STREAM_EXPIRATION_MEMBER,
},
],
);
expect(cacheStorageService.sortedSetRemove).not.toHaveBeenCalled();
});
it('does not track a stream whose key is missing during refresh', async () => {
cacheStorageService.expire
.mockResolvedValueOnce(false)
.mockResolvedValueOnce(true);
await expect(
service.refreshEventStreamTTL({
workspaceId: WORKSPACE_ID,
eventStreamChannelId: EVENT_STREAM_CHANNEL_ID,
}),
).resolves.toBe(false);
expect(cacheStorageService.sortedSetAdd).not.toHaveBeenCalled();
expect(cacheStorageService.sortedSetRemove).not.toHaveBeenCalled();
});
it('keeps a valid stream tracked when the workspace stream set is missing', async () => {
cacheStorageService.expire
.mockResolvedValueOnce(true)
.mockResolvedValueOnce(false);
await expect(
service.refreshEventStreamTTL({
workspaceId: WORKSPACE_ID,
eventStreamChannelId: EVENT_STREAM_CHANNEL_ID,
}),
).resolves.toBe(false);
expect(cacheStorageService.sortedSetAdd).toHaveBeenCalledWith(
ACTIVE_STREAM_EXPIRATIONS_KEY,
[
{
score: Date.now() + EVENT_STREAM_TTL_MS,
value: ACTIVE_STREAM_EXPIRATION_MEMBER,
},
],
);
expect(cacheStorageService.sortedSetRemove).not.toHaveBeenCalled();
});
it('removes a destroyed stream from its functional and metric indexes', async () => {
await service.destroyEventStream({
workspaceId: WORKSPACE_ID,
eventStreamChannelId: EVENT_STREAM_CHANNEL_ID,
});
expect(cacheStorageService.del).toHaveBeenCalledWith(EVENT_STREAM_KEY);
expect(cacheStorageService.setRemove).toHaveBeenCalledWith(
ACTIVE_STREAMS_KEY,
[EVENT_STREAM_CHANNEL_ID],
);
expect(cacheStorageService.sortedSetRemove).toHaveBeenCalledWith(
ACTIVE_STREAM_EXPIRATIONS_KEY,
[ACTIVE_STREAM_EXPIRATION_MEMBER],
);
});
it('removes stale streams from the functional and metric indexes', async () => {
await service.removeFromActiveStreams(WORKSPACE_ID, [
EVENT_STREAM_CHANNEL_ID,
'stale-stream-id',
]);
expect(cacheStorageService.setRemove).toHaveBeenCalledWith(
ACTIVE_STREAMS_KEY,
[EVENT_STREAM_CHANNEL_ID, 'stale-stream-id'],
);
expect(cacheStorageService.sortedSetRemove).toHaveBeenCalledWith(
ACTIVE_STREAM_EXPIRATIONS_KEY,
[ACTIVE_STREAM_EXPIRATION_MEMBER, `${WORKSPACE_ID}:stale-stream-id`],
);
});
});
packages/twenty-server/src/engine/subscriptions/event-stream.service.ts
+72 -2
View File
@@ -19,6 +19,7 @@ import {
} from 'src/engine/subscriptions/types/event-stream-data.type';
const ACTIVE_STREAM_COUNT_REFRESH_MS = 5 * 60 * 1_000;
const ACTIVE_STREAM_EXPIRATIONS_KEY = 'activeStreamExpirations';
@Injectable()
export class EventStreamService implements OnModuleInit {
@@ -50,8 +51,10 @@ export class EventStreamService implements OnModuleInit {
if (isStale) {
this.activeStreamCount =
await this.cacheStorageService.scanAndCountSetMembers(
'workspace:*:activeStreams',
await this.cacheStorageService.sortedSetRemoveByScoreAndCount(
ACTIVE_STREAM_EXPIRATIONS_KEY,
0,
now,
);
this.activeStreamCountRefreshedAt = now;
}
@@ -93,6 +96,11 @@ export class EventStreamService implements OnModuleInit {
[eventStreamChannelId],
EVENT_STREAM_TTL_MS,
);
await this.trackActiveStream({
workspaceId,
eventStreamChannelId,
});
}
async destroyEventStream({
@@ -110,6 +118,8 @@ export class EventStreamService implements OnModuleInit {
this.getActiveStreamsKey(workspaceId),
[eventStreamChannelId],
);
await this.untrackActiveStreams(workspaceId, [eventStreamChannelId]);
}
async getActiveStreamIds(workspaceId: string): Promise<string[]> {
@@ -130,6 +140,8 @@ export class EventStreamService implements OnModuleInit {
this.getActiveStreamsKey(workspaceId),
streamIdsToRemove,
);
await this.untrackActiveStreams(workspaceId, streamIdsToRemove);
}
async getStreamsData(
@@ -235,9 +247,67 @@ export class EventStreamService implements OnModuleInit {
this.cacheStorageService.expire(activeStreamsKey, EVENT_STREAM_TTL_MS),
]);
if (eventStreamRefreshed) {
await this.trackActiveStream({
workspaceId,
eventStreamChannelId,
});
}
return eventStreamRefreshed && activeStreamsRefreshed;
}
private async trackActiveStream({
workspaceId,
eventStreamChannelId,
}: {
workspaceId: string;
eventStreamChannelId: string;
}): Promise<void> {
try {
await this.cacheStorageService.sortedSetAdd(
ACTIVE_STREAM_EXPIRATIONS_KEY,
[
{
score: Date.now() + EVENT_STREAM_TTL_MS,
value: this.getActiveStreamExpirationMember(
workspaceId,
eventStreamChannelId,
),
},
],
);
} catch (error) {
this.logger.warn(`Failed to track active event stream: ${error}`);
}
}
private async untrackActiveStreams(
workspaceId: string,
eventStreamChannelIds: string[],
): Promise<void> {
try {
await this.cacheStorageService.sortedSetRemove(
ACTIVE_STREAM_EXPIRATIONS_KEY,
eventStreamChannelIds.map((eventStreamChannelId) =>
this.getActiveStreamExpirationMember(
workspaceId,
eventStreamChannelId,
),
),
);
} catch (error) {
this.logger.warn(`Failed to untrack active event streams: ${error}`);
}
}
private getActiveStreamExpirationMember(
workspaceId: string,
eventStreamChannelId: string,
): string {
return `${workspaceId}:${eventStreamChannelId}`;
}
private getEventStreamKey(
workspaceId: string,
eventStreamId: string,