feat: alternative sharded redis cache

packages/entity-cache-adapter-redis/src/GenericRedisCacher.ts

@@ -6,7 +6,6 @@ import {
   transformFieldsToCacheObject,
   IEntityGenericCacher,
 } from '@expo/entity';
-import { Redis } from 'ioredis';
 
 import { redisTransformerMap } from './RedisCommon';
 import wrapNativeRedisCallAsync from './errors/wrapNativeRedisCallAsync';
@@ -15,11 +14,22 @@ import wrapNativeRedisCallAsync from './errors/wrapNativeRedisCallAsync';
 // The sentinel value is distinct from any (positively) cached value.
 const DOES_NOT_EXIST_REDIS = '';
 
+export interface IRedisTransaction {
+  set(key: string, value: string, secondsToken: 'EX', seconds: number): this;
+  exec(): Promise<any>;
+}
+
+export interface IRedis {
+  mget(...args: [...keys: string[]]): Promise<(string | null)[]>;
+  multi(): IRedisTransaction;
+  del(...args: [...keys: string[]]): Promise<any>;
+}
+
 export interface GenericRedisCacheContext {
   /**
    * Instance of ioredis.Redis
    */
-  redisClient: Redis;
+  redisClient: IRedis;
 
   /**
    * TTL for caching database hits. Successive entity loads within this TTL

packages/entity-cache-adapter-redis/src/RedisCacheAdapter.ts

@@ -1,14 +1,13 @@
 import { EntityCacheAdapter, EntityConfiguration, CacheLoadResult, mapKeys } from '@expo/entity';
 import invariant from 'invariant';
-import type { Redis } from 'ioredis';
 
-import GenericRedisCacher from './GenericRedisCacher';
+import GenericRedisCacher, { IRedis } from './GenericRedisCacher';
 
 export interface RedisCacheAdapterContext {
   /**
    * Instance of ioredis.Redis
    */
-  redisClient: Redis;
+  redisClient: IRedis;
 
   /**
    * Create a key string for key parts (cache key prefix, versions, entity name, etc).

packages/entity-cache-adapter-redis/src/__integration-tests__/GenericRedisCacher-integration-test.ts

@@ -13,11 +13,12 @@ import { createRedisIntegrationTestEntityCompanionProvider } from '../testfixtur
 class TestViewerContext extends ViewerContext {}
 
 describe(GenericRedisCacher, () => {
+  const redisClient = new Redis(new URL(process.env['REDIS_URL']!).toString());
   let redisCacheAdapterContext: RedisCacheAdapterContext;
 
   beforeAll(() => {
     redisCacheAdapterContext = {
-      redisClient: new Redis(new URL(process.env['REDIS_URL']!).toString()),
+      redisClient,
       makeKeyFn(...parts: string[]): string {
         const delimiter = ':';
         const escapedParts = parts.map((part) =>
@@ -33,10 +34,10 @@ describe(GenericRedisCacher, () => {
   });
 
   beforeEach(async () => {
-    await redisCacheAdapterContext.redisClient.flushdb();
+    await redisClient.flushdb();
   });
   afterAll(async () => {
-    redisCacheAdapterContext.redisClient.disconnect();
+    redisClient.disconnect();
   });
 
   it('has correct caching and loading behavior', async () => {
@@ -62,7 +63,7 @@ describe(GenericRedisCacher, () => {
     ]);
     await genericRedisCacher.cacheManyAsync(objectMap);
 
-    const cachedJSON = await redisCacheAdapterContext.redisClient.get(testKey);
+    const cachedJSON = await redisClient.get(testKey);
     const cachedValue = JSON.parse(cachedJSON!);
     expect(cachedValue).toMatchObject({
       id: entity1Created.getID(),

packages/entity-cache-adapter-redis/src/__integration-tests__/RedisCacheAdapter-integration-test.ts

@@ -11,11 +11,12 @@ import { createRedisIntegrationTestEntityCompanionProvider } from '../testfixtur
 class TestViewerContext extends ViewerContext {}
 
 describe(RedisCacheAdapter, () => {
+  const redisClient = new Redis(new URL(process.env['REDIS_URL']!).toString());
   let redisCacheAdapterContext: RedisCacheAdapterContext;
 
   beforeAll(() => {
     redisCacheAdapterContext = {
-      redisClient: new Redis(new URL(process.env['REDIS_URL']!).toString()),
+      redisClient,
       makeKeyFn(...parts: string[]): string {
         const delimiter = ':';
         const escapedParts = parts.map((part) =>
@@ -31,10 +32,10 @@ describe(RedisCacheAdapter, () => {
   });
 
   beforeEach(async () => {
-    await redisCacheAdapterContext.redisClient.flushdb();
+    await redisClient.flushdb();
   });
   afterAll(async () => {
-    redisCacheAdapterContext.redisClient.disconnect();
+    redisClient.disconnect();
   });
 
   it('has correct caching behavior', async () => {
@@ -56,9 +57,7 @@ describe(RedisCacheAdapter, () => {
       .enforcing()
       .loadByIDAsync(entity1Created.getID());
 
-    const cachedJSON = await redisCacheAdapterContext.redisClient.get(
-      cacheKeyMaker('id', entity1.getID())
-    );
+    const cachedJSON = await redisClient.get(cacheKeyMaker('id', entity1.getID()));
     const cachedValue = JSON.parse(cachedJSON!);
     expect(cachedValue).toMatchObject({
       id: entity1.getID(),
@@ -73,9 +72,7 @@ describe(RedisCacheAdapter, () => {
     );
     expect(entityNonExistentResult.ok).toBe(false);
 
-    const nonExistentCachedValue = await redisCacheAdapterContext.redisClient.get(
-      cacheKeyMaker('id', nonExistentId)
-    );
+    const nonExistentCachedValue = await redisClient.get(cacheKeyMaker('id', nonExistentId));
     expect(nonExistentCachedValue).toEqual('');
 
     // load again through entities framework to ensure it reads negative result
@@ -86,9 +83,7 @@ describe(RedisCacheAdapter, () => {
 
     // invalidate from cache to ensure it invalidates correctly
     await RedisTestEntity.loader(viewerContext).invalidateFieldsAsync(entity1.getAllFields());
-    const cachedValueNull = await redisCacheAdapterContext.redisClient.get(
-      cacheKeyMaker('id', entity1.getID())
-    );
+    const cachedValueNull = await redisClient.get(cacheKeyMaker('id', entity1.getID()));
     expect(cachedValueNull).toBe(null);
   });
 

packages/entity-cache-adapter-redis/src/__integration-tests__/ShardedRedisCacheAdapter-integration-test.ts

@@ -0,0 +1,218 @@
+import { computeIfAbsent, mapMap, mapMapAsync, ViewerContext } from '@expo/entity';
+import invariant from 'invariant';
+import Redis from 'ioredis';
+import nullthrows from 'nullthrows';
+import { URL } from 'url';
+
+import { IRedis, IRedisTransaction } from '../GenericRedisCacher';
+import RedisCacheAdapter, { RedisCacheAdapterContext } from '../RedisCacheAdapter';
+import RedisTestEntity from '../testfixtures/RedisTestEntity';
+import { createRedisIntegrationTestEntityCompanionProvider } from '../testfixtures/createRedisIntegrationTestEntityCompanionProvider';
+
+//       shardingSchemeVersion: 1,
+
+const ids = {
+  entity1: '7345999c-b97d-47af-873c-bb1e9c1876f3', // shard 0
+  nonexistent: 'c207c238-70b5-41bb-8f33-130a57a6f6c5', // shard 1
+};
+
+function getShardForKey(key: string): number {
+  if (key.includes(ids.entity1)) {
+    return 0;
+  }
+
+  if (key.includes(ids.nonexistent)) {
+    return 1;
+  }
+
+  // entity1
+  if (key.includes('name:blah')) {
+    return 1; // put other field cache for entity1 on different server than the id field cached value
+  }
+
+  throw Error('unknown id encountered: ' + key);
+}
+
+function assertNotUndefined<T>(value: T | undefined): T {
+  invariant(value !== undefined, 'must not be undefined');
+  return value;
+}
+
+type SetInstruction = { key: string; value: string; secondsToken: 'EX'; seconds: number };
+
+class ShardedRedis implements IRedis {
+  constructor(private readonly redium: readonly Redis[]) {}
+
+  async mget(...args: string[]): Promise<(string | null)[]> {
+    const originalKeys = args;
+    const redisClientsForKeys = this.getRedisClientsForKeys(args);
+    const allKeysToValues = new Map<string, string | null>();
+    await mapMapAsync(redisClientsForKeys, async ({ redisClient, keys }) => {
+      const values = await redisClient.mget(keys);
+      for (let i = 0; i < keys.length; i++) {
+        allKeysToValues.set(nullthrows(keys[i]), assertNotUndefined(values[i]));
+      }
+    });
+
+    return originalKeys.map((k) => assertNotUndefined(allKeysToValues.get(k)));
+  }
+  multi(): IRedisTransaction {
+    const accumulatedSets: Map<string, SetInstruction> = new Map();
+    const parentExec = async (): Promise<void> => {
+      const redisClientsForKeys = this.getRedisClientsForKeys([...accumulatedSets.keys()]);
+      const redisClientsForInstructions = mapMap(redisClientsForKeys, ({ keys, redisClient }) => {
+        return {
+          keys,
+          sets: keys.map((k) => assertNotUndefined(accumulatedSets.get(k))),
+          redisClient,
+        };
+      });
+
+      await mapMapAsync(redisClientsForInstructions, async ({ sets, redisClient }) => {
+        let redisTransaction = redisClient.multi();
+        sets.forEach(({ key, value, secondsToken, seconds }) => {
+          redisTransaction = redisTransaction.set(key, value, secondsToken, seconds);
+        });
+        await redisTransaction.exec();
+      });
+    };
+    const ret: IRedisTransaction = {
+      set(key, value, secondsToken, seconds): any {
+        accumulatedSets.set(key, { key, value, secondsToken, seconds });
+        return ret;
+      },
+      async exec(): Promise<any> {
+        return await parentExec();
+      },
+    };
+    return ret;
+  }
+
+  async del(...args: string[]): Promise<void> {
+    const redisClientsForKeys = this.getRedisClientsForKeys(args);
+    await mapMapAsync(redisClientsForKeys, async ({ keys, redisClient }) => {
+      await redisClient.del(...keys);
+    });
+  }
+
+  private getRedisClientsForKeys(
+    keys: readonly string[]
+  ): Map<number, { keys: string[]; redisClient: Redis }> {
+    const shardGroupsForKeys = new Map(keys.map((k) => [k, getShardForKey(k)]));
+    const redisClientsForKeys = new Map<number, { keys: string[]; redisClient: Redis }>();
+    for (const [key, shardGroup] of shardGroupsForKeys) {
+      const entry = computeIfAbsent(redisClientsForKeys, shardGroup, (currShardGroup) => ({
+        keys: [],
+        redisClient: this.getRedisInstanceForShardGroup(currShardGroup),
+      }));
+      entry.keys.push(key);
+    }
+    return redisClientsForKeys;
+  }
+
+  private getRedisInstanceForShardGroup(shardGroup: number): Redis {
+    return nullthrows(this.redium[shardGroup]);
+  }
+}
+
+describe(RedisCacheAdapter, () => {
+  const redium = [
+    new Redis(new URL(process.env['REDIS_URL']!).toString()),
+    new Redis(new URL(process.env['REDIS_URL_2']!).toString()),
+  ];
+
+  const redisClient = new ShardedRedis(redium);
+
+  let redisCacheAdapterContext: RedisCacheAdapterContext;
+
+  beforeAll(() => {
+    redisCacheAdapterContext = {
+      redisClient,
+      makeKeyFn(...parts: string[]): string {
+        const delimiter = ':';
+        const escapedParts = parts.map((part) =>
+          part.replace('\\', '\\\\').replace(delimiter, `\\${delimiter}`)
+        );
+        return escapedParts.join(delimiter);
+      },
+      cacheKeyPrefix: 'test-',
+      cacheKeyVersion: 1,
+      ttlSecondsPositive: 86400, // 1 day
+      ttlSecondsNegative: 600, // 10 minutes
+    };
+  });
+
+  beforeEach(async () => {
+    await Promise.all(redium.map((redis) => redis.flushdb()));
+  });
+
+  afterAll(async () => {
+    redium.map((redis) => redis.disconnect());
+  });
+
+  it('has correct caching behavior', async () => {
+    const viewerContext = new ViewerContext(
+      createRedisIntegrationTestEntityCompanionProvider(redisCacheAdapterContext)
+    );
+    const cacheAdapter = viewerContext.entityCompanionProvider.getCompanionForEntity(
+      RedisTestEntity,
+      RedisTestEntity.getCompanionDefinition()
+    )['tableDataCoordinator']['cacheAdapter'];
+    const cacheKeyMaker = cacheAdapter['makeCacheKey'].bind(cacheAdapter);
+    const entity1Created = await RedisTestEntity.creator(viewerContext)
+      .setField('id', ids.entity1)
+      .setField('name', 'blah')
+      .enforceCreateAsync();
+
+    // loading an entity should put it in cache. load by both cached fields (which are on separate redis instances).
+    const cacheKeyEntity1 = cacheKeyMaker('id', entity1Created.getID());
+    const cacheKeyEntity1ShardGroup = getShardForKey(cacheKeyEntity1);
+    const cacheRedisClientForEntity = nullthrows(redium[cacheKeyEntity1ShardGroup]);
+    const entity1 = await RedisTestEntity.loader(viewerContext)
+      .enforcing()
+      .loadByIDAsync(entity1Created.getID());
+    const cachedJSON = await cacheRedisClientForEntity.get(cacheKeyEntity1);
+    const cachedValue = JSON.parse(cachedJSON!);
+    expect(cachedValue).toMatchObject({
+      id: entity1.getID(),
+      name: 'blah',
+    });
+
+    const cacheKeyEntity1NameField = cacheKeyMaker('name', entity1Created.getField('name'));
+    const cacheKeyEntity1NameFieldShardGroup = getShardForKey(cacheKeyEntity1NameField);
+    const cacheRedisClientForEntityNameField = nullthrows(
+      redium[cacheKeyEntity1NameFieldShardGroup]
+    );
+    await RedisTestEntity.loader(viewerContext)
+      .enforcing()
+      .loadByFieldEqualingAsync('name', entity1Created.getField('name'));
+    await expect(cacheRedisClientForEntityNameField.get(cacheKeyEntity1NameField)).resolves.toEqual(
+      cachedJSON
+    );
+
+    expect(cacheKeyEntity1ShardGroup).not.toEqual(cacheKeyEntity1NameFieldShardGroup);
+
+    // simulate non existent db fetch, should write negative result ('') to cache
+    const nonExistentId = ids.nonexistent;
+    const entityNonExistentResult = await RedisTestEntity.loader(viewerContext).loadByIDAsync(
+      nonExistentId
+    );
+    expect(entityNonExistentResult.ok).toBe(false);
+    const cacheKeyNonExistent = cacheKeyMaker('id', nonExistentId);
+    const cacheRedisClientForNonExistent = nullthrows(redium[getShardForKey(cacheKeyNonExistent)]);
+    const nonExistentCachedValue = await cacheRedisClientForNonExistent.get(cacheKeyNonExistent);
+    expect(nonExistentCachedValue).toEqual('');
+    // load again through entities framework to ensure it reads negative result
+    const entityNonExistentResult2 = await RedisTestEntity.loader(viewerContext).loadByIDAsync(
+      nonExistentId
+    );
+    expect(entityNonExistentResult2.ok).toBe(false);
+
+    // invalidate from cache to ensure it invalidates correctly in both caches
+    await RedisTestEntity.loader(viewerContext).invalidateFieldsAsync(entity1.getAllFields());
+    await expect(cacheRedisClientForEntity.get(cacheKeyEntity1)).resolves.toBeNull();
+    await expect(
+      cacheRedisClientForEntityNameField.get(cacheKeyEntity1NameField)
+    ).resolves.toBeNull();
+  });
+});

packages/entity-cache-adapter-redis/src/__integration-tests__/errors-test.ts

@@ -9,11 +9,12 @@ import { createRedisIntegrationTestEntityCompanionProvider } from '../testfixtur
 class TestViewerContext extends ViewerContext {}
 
 describe(RedisCacheAdapter, () => {
+  const redisClient = new Redis(new URL(process.env['REDIS_URL']!).toString());
   let redisCacheAdapterContext: RedisCacheAdapterContext;
 
   beforeAll(() => {
     redisCacheAdapterContext = {
-      redisClient: new Redis(new URL(process.env['REDIS_URL']!).toString()),
+      redisClient,
       makeKeyFn(...parts: string[]): string {
         const delimiter = ':';
         const escapedParts = parts.map((part) =>
@@ -29,11 +30,11 @@ describe(RedisCacheAdapter, () => {
   });
 
   beforeEach(async () => {
-    await redisCacheAdapterContext.redisClient.flushdb();
+    await redisClient.flushdb();
   });
 
   it('throws when redis is disconnected', async () => {
-    redisCacheAdapterContext.redisClient.disconnect();
+    redisClient.disconnect();
 
     const vc1 = new TestViewerContext(
       createRedisIntegrationTestEntityCompanionProvider(redisCacheAdapterContext)

packages/entity-full-integration-tests/src/__integration-tests__/EntityCacheInconsistency-test.ts

@@ -97,6 +97,7 @@ async function dropPostgresTable(knex: Knex): Promise<void> {
 
 describe('Entity cache inconsistency', () => {
   let knexInstance: Knex;
+  const redisClient = new Redis(new URL(process.env['REDIS_URL']!).toString());
   let redisCacheAdapterContext: RedisCacheAdapterContext;
 
   beforeAll(() => {
@@ -111,7 +112,7 @@ describe('Entity cache inconsistency', () => {
       },
     });
     redisCacheAdapterContext = {
-      redisClient: new Redis(new URL(process.env['REDIS_URL']!).toString()),
+      redisClient,
       makeKeyFn(...parts: string[]): string {
         const delimiter = ':';
         const escapedParts = parts.map((part) =>
@@ -128,13 +129,13 @@ describe('Entity cache inconsistency', () => {
 
   beforeEach(async () => {
     await createOrTruncatePostgresTables(knexInstance);
-    await redisCacheAdapterContext.redisClient.flushdb();
+    await redisClient.flushdb();
   });
 
   afterAll(async () => {
     await dropPostgresTable(knexInstance);
     await knexInstance.destroy();
-    redisCacheAdapterContext.redisClient.disconnect();
+    redisClient.disconnect();
   });
 
   test('lots of updates in long-ish running transactions', async () => {

packages/entity-full-integration-tests/src/__integration-tests__/EntityEdgesIntegration-test.ts

@@ -33,6 +33,7 @@ async function dropPostgresTable(knex: Knex): Promise<void> {
 
 describe('EntityMutator.processEntityDeletionForInboundEdgesAsync', () => {
   let knexInstance: Knex;
+  const redisClient = new Redis(new URL(process.env['REDIS_URL']!).toString());
   let redisCacheAdapterContext: RedisCacheAdapterContext;
 
   beforeAll(() => {
@@ -47,7 +48,7 @@ describe('EntityMutator.processEntityDeletionForInboundEdgesAsync', () => {
       },
     });
     redisCacheAdapterContext = {
-      redisClient: new Redis(new URL(process.env['REDIS_URL']!).toString()),
+      redisClient,
       makeKeyFn(...parts: string[]): string {
         const delimiter = ':';
         const escapedParts = parts.map((part) =>
@@ -64,13 +65,13 @@ describe('EntityMutator.processEntityDeletionForInboundEdgesAsync', () => {
 
   beforeEach(async () => {
     await createOrTruncatePostgresTables(knexInstance);
-    await redisCacheAdapterContext.redisClient.flushdb();
+    await redisClient.flushdb();
   });
 
   afterAll(async () => {
     await dropPostgresTable(knexInstance);
     await knexInstance.destroy();
-    redisCacheAdapterContext.redisClient.disconnect();
+    redisClient.disconnect();
   });
 
   describe('EntityEdgeDeletionBehavior.INVALIDATE_CACHE', () => {

packages/entity-full-integration-tests/src/__integration-tests__/EntitySelfReferentialEdgesIntegration-test.ts

@@ -176,6 +176,7 @@ const makeEntityClasses = async (knex: Knex, edgeDeletionBehavior: EntityEdgeDel
 };
 describe('EntityMutator.processEntityDeletionForInboundEdgesAsync', () => {
   let knexInstance: Knex;
+  const redisClient = new Redis(new URL(process.env['REDIS_URL']!).toString());
   let redisCacheAdapterContext: RedisCacheAdapterContext;
 
   beforeAll(() => {
@@ -190,7 +191,7 @@ describe('EntityMutator.processEntityDeletionForInboundEdgesAsync', () => {
       },
     });
     redisCacheAdapterContext = {
-      redisClient: new Redis(new URL(process.env['REDIS_URL']!).toString()),
+      redisClient,
       makeKeyFn(...parts: string[]): string {
         const delimiter = ':';
         const escapedParts = parts.map((part) =>
@@ -207,7 +208,7 @@ describe('EntityMutator.processEntityDeletionForInboundEdgesAsync', () => {
 
   afterAll(async () => {
     await knexInstance.destroy();
-    redisCacheAdapterContext.redisClient.disconnect();
+    redisClient.disconnect();
   });
 
   it.each([

packages/entity-secondary-cache-redis/src/__integration-tests__/RedisSecondaryEntityCache-integration-test.ts

@@ -52,11 +52,12 @@ class TestSecondaryRedisCacheLoader extends EntitySecondaryCacheLoader<
 }
 
 describe(RedisSecondaryEntityCache, () => {
+  const redisClient = new Redis(new URL(process.env['REDIS_URL']!).toString());
   let redisCacheAdapterContext: RedisCacheAdapterContext;
 
   beforeAll(() => {
     redisCacheAdapterContext = {
-      redisClient: new Redis(new URL(process.env['REDIS_URL']!).toString()),
+      redisClient,
       makeKeyFn(..._parts: string[]): string {
         throw new Error('should not be used by this test');
       },
@@ -68,10 +69,10 @@ describe(RedisSecondaryEntityCache, () => {
   });
 
   beforeEach(async () => {
-    await redisCacheAdapterContext.redisClient.flushdb();
+    await redisClient.flushdb();
   });
   afterAll(async () => {
-    redisCacheAdapterContext.redisClient.disconnect();
+    redisClient.disconnect();
   });
 
   it('Loads through secondary loader, caches, and invalidates', async () => {

resources/docker-compose.yml

@@ -6,6 +6,12 @@ services:
       - "${REDIS_PORT}:6379"
     volumes:
       - redis_data:/data
+  redis2:
+    image: redis
+    ports:
+      - "${REDIS_PORT_2}:6379"
+    volumes:
+      - redis_data_2:/data
   postgres:
     image: postgres:14
     ports:
@@ -20,3 +26,4 @@ services:
 volumes:
   postgres_data: {}
   redis_data: {}
+  redis_data_2: {}

resources/test-env.sh

@@ -3,7 +3,9 @@
 export NODE_ENV=test
 
 export REDIS_PORT="16379"
+export REDIS_PORT_2="17379"
 export REDIS_URL="redis://localhost:${REDIS_PORT}/0"
+export REDIS_URL_2="redis://localhost:${REDIS_PORT_2}/0"
 
 export PGUSER="postgresuser"
 export PGPASSWORD="handkerchief-break-popular-population"