fix(server): stop cross-pod recompute cascade on localDataOnly workspace cache keys (#22980)

## Context Prod investigation (Sentry, last 7 days) traced the current slowness to the per-pod workspace cache. Every recompute of a `localDataOnly` key (`ORMEntityMetadatas`, `flatWorkspaceMemberMaps`) published a fresh `crypto.randomUUID()` as the shared Redis validation hash. Because these keys recover from a hash mismatch by recomputing (their data never enters Redis), one miss on one pod invalidated the local copy on every other pod; each of their recomputes minted yet another hash, re-invalidating everyone else. The fleet never converges. Measured impact in prod: - The `ORMEntityMetadatas` rebuild (full `objectMetadata` + `fieldMetadata` + `application` queries, ~220ms combined, plus `EntityMetadataBuilder.build`) ran **~963k times in 24h** (~11/s), roughly 58h of cumulative Postgres time per day. - The hottest single workspace recomputed its schema metadata 51k times/day (once per 1.7s). - Second-order effects: `POST /metadata` averaged 26.6s (p95 2.3s, so a tail hangs for minutes on pool/event-loop starvation), GraphQL p95 went 846ms (v2.20.0) to 1744ms (v2.21.0), `Query read timeout` on trivial cron queries at 18x baseline. The random hash was correct in the original design (#15962): it is a generation token, and Redis-backed keys recover absorptively by adopting hash+data from Redis. #16287 added `localOnly` keys (EntityMetadata[] is not serializable) whose recovery is generative, which silently broke the invariant later documented in #18649 ("hashes change only on invalidateAndRecompute"). ## What this does - Recovery recomputes now **adopt** the hash already present in Redis instead of minting a new one, and write nothing back. A miss costs one recompute on one pod instead of an unbounded fleet-wide loop. - Minting is reserved for `invalidateAndRecompute` (real metadata changes, propagation semantics unchanged, including the frontend collectionHashes contract) and the bootstrap case where Redis has no hash. - The bootstrap write uses **SET NX** (new `CacheStorageService.setIfAbsent`) instead of a plain overwrite: a slow bootstrap recompute could otherwise land after a concurrent `invalidateAndRecompute` mint and clobber it with a hash of pre-migration data. Under the old code that clobber self-healed via the cascade; with adopt semantics it would pin stale data, so the bootstrap write must lose that race. A losing pod keeps its result locally as provisional and converges on the winning hash at its next revalidation (covered by a dedicated race test). Redis-backed keys are untouched: same fetch-on-mismatch recovery, same mint-and-write on `missingInRedis`. ## Expected effect and how to verify `FieldMetadataEntity`/`ObjectMetadataEntity`/`ApplicationEntity` full-workspace query counts in Sentry should collapse from ~1M/day to the true metadata-change rate, and with them the DB pool pressure behind the `/metadata` latency tail. This also makes local-cache eviction (`MAX_LOCAL_CACHE_ENTRIES`, #22946) cheap: the cap can be tuned purely for RAM. Complementary to, not competing with, the planned Redis pub/sub invalidation: a version token in Redis is still needed for restart catch-up, and this PR gives it sound semantics. --- _Generated by [Claude Code](https://claude.ai/code/session_01T3JUHwXJHPmZDZTrv6YTDi)_  <a href="https://cubic.dev/pr/twentyhq/twenty/pull/22980?utm_source=github" target="_blank" rel="noopener noreferrer" data-no-image-dialog="true"><picture><source media="(prefers-color-scheme: dark)" srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source media="(prefers-color-scheme: light)" srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img alt="Review in cubic" src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a>
2026-07-20 15:26:49 +02:00
parent e6c6cccafa
commit d4ac6e752b
3 changed files with 262 additions and 11 deletions
@@ -24,6 +24,32 @@ export class CacheStorageService {
    return this.cache.set(this.getKey(key), value, ttl);
  }

+  async setIfAbsent<T>(
+    key: string,
+    value: T,
+    ttl: Milliseconds,
+  ): Promise<boolean> {
+    if (this.isRedisCache()) {
+      const result = await (this.cache as RedisCache).store.client.set(
+        this.getKey(key),
+        JSON.stringify(value),
+        ttl > 0 ? { NX: true, PX: ttl } : { NX: true },
+      );
+
+      return result === 'OK';
+    }
+
+    const existingValue = await this.get(key);
+
+    if (existingValue !== undefined) {
+      return false;
+    }
+
+    await this.set(key, value, ttl);
+
+    return true;
+  }
+
  async del(key: string) {
    return this.cache.del(this.getKey(key));
  }
@@ -6,7 +6,11 @@ import { WorkspaceCacheProvider } from 'src/engine/workspace-cache/interfaces/wo
 import { type CacheStorageService } from 'src/engine/core-modules/cache-storage/services/cache-storage.service';
 import { CacheStorageNamespace } from 'src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum';
 import { MetricsService } from 'src/engine/core-modules/metrics/metrics.service';
-import { WORKSPACE_CACHE_KEY } from 'src/engine/workspace-cache/decorators/workspace-cache.decorator';
+import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';
+import {
+  WORKSPACE_CACHE_KEY,
+  WORKSPACE_CACHE_OPTIONS,
+} from 'src/engine/workspace-cache/decorators/workspace-cache.decorator';
 import { WorkspaceCacheService } from 'src/engine/workspace-cache/services/workspace-cache.service';

 const WORKSPACE_ID = '20202020-0000-4000-8000-000000000000';
@@ -27,6 +31,14 @@ class MockRolesPermissionsCacheProvider extends WorkspaceCacheProvider<{
  }
 }

+class MockOrmEntityMetadatasCacheProvider extends WorkspaceCacheProvider<{
+  testData: string;
+}> {
+  async computeForCache(_workspaceId: string) {
+    return { testData: 'orm-computed-value' };
+  }
+}
+
 describe('WorkspaceCacheService', () => {
  let service: WorkspaceCacheService;
  let cacheStorageService: jest.Mocked<CacheStorageService>;
@@ -48,6 +60,7 @@ describe('WorkspaceCacheService', () => {
            mget: jest.fn(),
            mset: jest.fn(),
            mdel: jest.fn(),
+            setIfAbsent: jest.fn(),
          },
        },
        {
@@ -68,6 +81,12 @@ describe('WorkspaceCacheService', () => {
            incrementCounterBy: jest.fn(),
          },
        },
+        {
+          provide: TwentyConfigService,
+          useValue: {
+            get: jest.fn().mockReturnValue(604800),
+          },
+        },
      ],
    }).compile();

@@ -510,4 +529,145 @@ describe('WorkspaceCacheService', () => {
      expect(result).toEqual({ featureFlagsMap: { testData: 'latest-value' } });
    });
  });
+
+  describe('localDataOnly hash recovery', () => {
+    let localDataOnlyProvider: MockOrmEntityMetadatasCacheProvider;
+    const ormHashKey = `orm:entity-metadatas:${WORKSPACE_ID}:hash`;
+    const ormDataKey = `orm:entity-metadatas:${WORKSPACE_ID}:data`;
+
+    beforeEach(async () => {
+      localDataOnlyProvider = new MockOrmEntityMetadatasCacheProvider();
+
+      discoveryService.getProviders.mockReturnValue([
+        { instance: localDataOnlyProvider },
+      ] as any);
+
+      reflector.get.mockImplementation((key, target) => {
+        if (target !== MockOrmEntityMetadatasCacheProvider) {
+          return undefined;
+        }
+
+        if (key === WORKSPACE_CACHE_KEY) {
+          return 'ORMEntityMetadatas';
+        }
+
+        if (key === WORKSPACE_CACHE_OPTIONS) {
+          return { localDataOnly: true };
+        }
+
+        return undefined;
+      });
+
+      await service.onModuleInit();
+    });
+
+    it('should adopt the existing redis hash when recomputing instead of minting a new one', async () => {
+      cacheStorageService.mget.mockResolvedValue(['hash-from-another-pod']);
+      cacheStorageService.mset.mockResolvedValue(undefined);
+
+      const computeSpy = jest.spyOn(localDataOnlyProvider, 'computeForCache');
+
+      const result = await service.getOrRecompute(WORKSPACE_ID, [
+        'ORMEntityMetadatas',
+      ]);
+
+      expect(result).toEqual({
+        ORMEntityMetadatas: { testData: 'orm-computed-value' },
+      });
+      expect(computeSpy).toHaveBeenCalledTimes(1);
+      expect(cacheStorageService.mset).not.toHaveBeenCalled();
+    });
+
+    it('should treat the local copy as valid on later reads after adopting the redis hash', async () => {
+      cacheStorageService.mget.mockResolvedValue(['hash-from-another-pod']);
+      cacheStorageService.mset.mockResolvedValue(undefined);
+
+      const computeSpy = jest.spyOn(localDataOnlyProvider, 'computeForCache');
+
+      await service.getOrRecompute(WORKSPACE_ID, ['ORMEntityMetadatas']);
+
+      jest.advanceTimersByTime(15_000);
+
+      await service.getOrRecompute(WORKSPACE_ID, ['ORMEntityMetadatas']);
+
+      expect(computeSpy).toHaveBeenCalledTimes(1);
+      expect(cacheStorageService.mset).not.toHaveBeenCalled();
+    });
+
+    it('should recompute and adopt the new hash when another pod invalidated the key', async () => {
+      cacheStorageService.mget.mockResolvedValue(['hash-v1']);
+      cacheStorageService.mset.mockResolvedValue(undefined);
+
+      const computeSpy = jest.spyOn(localDataOnlyProvider, 'computeForCache');
+
+      await service.getOrRecompute(WORKSPACE_ID, ['ORMEntityMetadatas']);
+
+      jest.advanceTimersByTime(15_000);
+
+      cacheStorageService.mget.mockResolvedValue(['hash-v2']);
+
+      await service.getOrRecompute(WORKSPACE_ID, ['ORMEntityMetadatas']);
+
+      expect(computeSpy).toHaveBeenCalledTimes(2);
+      expect(cacheStorageService.mset).not.toHaveBeenCalled();
+    });
+
+    it('should mint a hash and persist it only if still absent when redis has none', async () => {
+      cacheStorageService.mget.mockResolvedValue([undefined]);
+      cacheStorageService.setIfAbsent.mockResolvedValue(true);
+
+      await service.getOrRecompute(WORKSPACE_ID, ['ORMEntityMetadatas']);
+
+      expect(cacheStorageService.setIfAbsent).toHaveBeenCalledWith(
+        ormHashKey,
+        expect.any(String),
+        604800000,
+      );
+      expect(cacheStorageService.mset).not.toHaveBeenCalled();
+    });
+
+    it('should converge on a hash written concurrently by another pod instead of overwriting it', async () => {
+      cacheStorageService.mget.mockResolvedValue([undefined]);
+      cacheStorageService.setIfAbsent.mockResolvedValue(false);
+
+      const computeSpy = jest.spyOn(localDataOnlyProvider, 'computeForCache');
+
+      await service.getOrRecompute(WORKSPACE_ID, ['ORMEntityMetadatas']);
+
+      expect(cacheStorageService.setIfAbsent).toHaveBeenCalledTimes(1);
+      expect(cacheStorageService.mset).not.toHaveBeenCalled();
+
+      jest.advanceTimersByTime(15_000);
+
+      cacheStorageService.mget.mockResolvedValue(['winner-hash']);
+
+      await service.getOrRecompute(WORKSPACE_ID, ['ORMEntityMetadatas']);
+
+      expect(computeSpy).toHaveBeenCalledTimes(2);
+      expect(cacheStorageService.setIfAbsent).toHaveBeenCalledTimes(1);
+
+      jest.advanceTimersByTime(15_000);
+
+      await service.getOrRecompute(WORKSPACE_ID, ['ORMEntityMetadatas']);
+
+      expect(computeSpy).toHaveBeenCalledTimes(2);
+    });
+
+    it('should mint a new hash on invalidateAndRecompute', async () => {
+      cacheStorageService.mdel.mockResolvedValue(undefined);
+      cacheStorageService.mset.mockResolvedValue(undefined);
+
+      await service.invalidateAndRecompute(WORKSPACE_ID, [
+        'ORMEntityMetadatas',
+      ]);
+
+      expect(cacheStorageService.mdel).toHaveBeenCalledWith([
+        ormDataKey,
+        ormHashKey,
+      ]);
+      expect(cacheStorageService.mset).toHaveBeenCalledWith([
+        { key: ormHashKey, value: expect.any(String) },
+      ]);
+    });
+  });
 });
@@ -12,6 +12,7 @@ import { CacheStorageService } from 'src/engine/core-modules/cache-storage/servi
 import { CacheStorageNamespace } from 'src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum';
 import { MetricsService } from 'src/engine/core-modules/metrics/metrics.service';
 import { MetricsKeys } from 'src/engine/core-modules/metrics/types/metrics-keys.type';
+import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';
 import { PromiseMemoizer } from 'src/engine/twenty-orm/storage/promise-memoizer.storage';
 import {
  WORKSPACE_CACHE_KEY,
@@ -41,6 +42,13 @@ const MIN_EVICT_KEYS = 100;

 type CacheDataType = WorkspaceCacheDataMap[WorkspaceCacheKeyName];

+type RecomputeHashResolution =
+  | { strategy: 'mint' }
+  | {
+      strategy: 'recover';
+      adoptableHashes: Partial<Record<WorkspaceCacheKeyName, string>>;
+    };
+
@Injectable()
 export class WorkspaceCacheService implements OnModuleInit {
  private readonly localCache = new Map<
@@ -64,6 +72,7 @@ export class WorkspaceCacheService implements OnModuleInit {
    private readonly discoveryService: DiscoveryService,
    private readonly reflector: Reflector,
    private readonly metricsService: MetricsService,
+    private readonly twentyConfigService: TwentyConfigService,
  ) {}

  async onModuleInit() {
@@ -135,8 +144,15 @@ export class WorkspaceCacheService implements OnModuleInit {
        }

        // Stage 2: Validate ttl stale keys against Redis hash
-        const { validKeys, keysNeedingDataFromRedis, keysNeedingRecompute } =
-          await this.validateLocalHashAgainstRedisHash(workspaceId, staleKeys);
+        const {
+          validKeys,
+          keysNeedingDataFromRedis,
+          keysNeedingRecompute,
+          adoptableHashes,
+        } = await this.validateLocalHashAgainstRedisHash(
+          workspaceId,
+          staleKeys,
+        );
        const validatedData = this.getFromLocalCache(workspaceId, validKeys);

        // Stage 3: Fetch data from Redis
@@ -150,6 +166,7 @@ export class WorkspaceCacheService implements OnModuleInit {
        const recomputedData = await this.recomputeDataFromProvider(
          workspaceId,
          keysToRecompute,
+          { strategy: 'recover', adoptableHashes },
        );

        return {
@@ -171,7 +188,9 @@ export class WorkspaceCacheService implements OnModuleInit {
    await this.memoizer.clearKeys(`${workspaceId}-`);

    await this.flush(workspaceId, cacheKeyNames);
-    await this.recomputeDataFromProvider(workspaceId, cacheKeyNames);
+    await this.recomputeDataFromProvider(workspaceId, cacheKeyNames, {
+      strategy: 'mint',
+    });

    // Clear memoizer again after recomputation to evict any stale entries
    // cached by concurrent getOrRecompute calls during the flush window.
@@ -241,13 +260,20 @@ export class WorkspaceCacheService implements OnModuleInit {
    validKeys: WorkspaceCacheKeyName[];
    keysNeedingDataFromRedis: WorkspaceCacheKeyName[];
    keysNeedingRecompute: WorkspaceCacheKeyName[];
+    adoptableHashes: Partial<Record<WorkspaceCacheKeyName, string>>;
  }> {
    const validKeys: WorkspaceCacheKeyName[] = [];
    const keysNeedingDataFromRedis: WorkspaceCacheKeyName[] = [];
    const keysNeedingRecompute: WorkspaceCacheKeyName[] = [];
+    const adoptableHashes: Partial<Record<WorkspaceCacheKeyName, string>> = {};

    if (cacheKeyNames.length === 0) {
-      return { validKeys, keysNeedingDataFromRedis, keysNeedingRecompute };
+      return {
+        validKeys,
+        keysNeedingDataFromRedis,
+        keysNeedingRecompute,
+        adoptableHashes,
+      };
    }

    const hashKeys = cacheKeyNames.map(
@@ -270,12 +296,21 @@ export class WorkspaceCacheService implements OnModuleInit {
        validKeys.push(keyName);
      } else if (this.localDataOnlyKeys.has(keyName)) {
        keysNeedingRecompute.push(keyName);
+
+        if (isDefined(redisHash)) {
+          adoptableHashes[keyName] = redisHash;
+        }
      } else {
        keysNeedingDataFromRedis.push(keyName);
      }
    }

-    return { validKeys, keysNeedingDataFromRedis, keysNeedingRecompute };
+    return {
+      validKeys,
+      keysNeedingDataFromRedis,
+      keysNeedingRecompute,
+      adoptableHashes,
+    };
  }

  private async fetchDataFromRedis(
@@ -321,6 +356,7 @@ export class WorkspaceCacheService implements OnModuleInit {
  private async recomputeDataFromProvider(
    workspaceId: string,
    cacheKeyNames: WorkspaceCacheKeyName[],
+    hashResolution: RecomputeHashResolution,
  ): Promise<Partial<WorkspaceCacheDataMap>> {
    const result: Partial<WorkspaceCacheDataMap> = {};

@@ -331,23 +367,41 @@ export class WorkspaceCacheService implements OnModuleInit {
    const computePromises = cacheKeyNames.map(async (keyName) => {
      const provider = this.getProviderOrThrow(keyName);
      const data = await provider.computeForCache(workspaceId);
-      const hash = crypto.randomUUID();

-      return { keyName, data, hash };
+      if (hashResolution.strategy === 'mint') {
+        return { keyName, data, hash: crypto.randomUUID(), isAdopted: false };
+      }
+
+      const adoptableHash = hashResolution.adoptableHashes[keyName];
+
+      return {
+        keyName,
+        data,
+        hash: adoptableHash ?? crypto.randomUUID(),
+        isAdopted: isDefined(adoptableHash),
+      };
    });

    const computed = await Promise.all(computePromises);

    const redisEntries: Array<{ key: string; value: unknown }> = [];
+    const bootstrapHashEntries: Array<{ key: string; value: string }> = [];

-    for (const { keyName, data, hash } of computed) {
+    for (const { keyName, data, hash, isAdopted } of computed) {
      Object.assign(result, { [keyName]: data });

      const baseKey = this.buildCacheKey(workspaceId, keyName);
+      const isLocalDataOnly = this.localDataOnlyKeys.has(keyName);
+      const isRecoveryBootstrap =
+        hashResolution.strategy === 'recover' && !isAdopted && isLocalDataOnly;

-      redisEntries.push({ key: `${baseKey}:hash`, value: hash });
+      if (isRecoveryBootstrap) {
+        bootstrapHashEntries.push({ key: `${baseKey}:hash`, value: hash });
+      } else if (!isAdopted) {
+        redisEntries.push({ key: `${baseKey}:hash`, value: hash });
+      }

-      if (!this.localDataOnlyKeys.has(keyName)) {
+      if (!isLocalDataOnly) {
        redisEntries.push({ key: `${baseKey}:data`, value: data });
      }

@@ -358,6 +412,17 @@ export class WorkspaceCacheService implements OnModuleInit {
      await this.cacheStorage.mset(redisEntries);
    }

+    if (bootstrapHashEntries.length > 0) {
+      const bootstrapHashTtlMs =
+        this.twentyConfigService.get('CACHE_STORAGE_TTL') * 1000;
+
+      await Promise.all(
+        bootstrapHashEntries.map(({ key, value }) =>
+          this.cacheStorage.setIfAbsent(key, value, bootstrapHashTtlMs),
+        ),
+      );
+    }
+
    return result;
  }