wip

Author: sjpotter

package-lock.json

@@ -8,9 +8,9 @@ type CachingClientType = RedisClientType<any, any, any, any, any>;
 type CmdFunc = () => Promise<ReplyUnion>;
 
 export interface ClientSideCacheConfig {
-  ttl: number;
-  maxEntries: number;
-  lru: boolean;
+  ttl?: number;
+  maxEntries?: number;
+  lru?: boolean;
 }
 
 type CacheCreator = {
@@ -48,7 +48,7 @@ export class ClientSideCacheEntryValue extends ClientSideCacheEntryBase {
   readonly #value: any;
 
   get value() {
-    return structuredClone(this.#value);
+    return this.#value;
   }
 
   constructor(ttl: number, value: any) {
@@ -79,7 +79,6 @@ export abstract class ClientSideCacheProvider extends EventEmitter {
   abstract cacheMisses(): number;
   abstract onError(): void;
   abstract onClose(): void;
-  abstract onDestroy(): void;
 }
 
 export class BasicClientSideCache extends ClientSideCacheProvider {
@@ -98,7 +97,7 @@ export class BasicClientSideCache extends ClientSideCacheProvider {
     this.#keyToCacheKeySetMap = new Map<string, Set<string>>();
     this.ttl = config?.ttl ?? 0;
     this.#maxEntries = config?.maxEntries ?? 0;
-    this.#lru = config?.lru ?? false;
+    this.#lru = config?.lru ?? true;
   }
 
   /* logic of how caching works:
@@ -123,7 +122,6 @@ export class BasicClientSideCache extends ClientSideCacheProvider {
     transformReply: TransformReply | undefined,
     typeMapping: TypeMapping | undefined
   ) {
-    console.log("handleCache: enter");
     let reply: ReplyUnion;
 
     const cacheKey = parser.cacheKey;
@@ -133,10 +131,9 @@ export class BasicClientSideCache extends ClientSideCacheProvider {
     if (cacheEntry) {
       // If instanceof is "too slow", can add a "type" and then use an "as" cast to call proper getters.
       if (cacheEntry instanceof ClientSideCacheEntryValue) { // "2b1"
-        console.log("returning value from cache");
         this.#cacheHit();
 
-        return cacheEntry.value;
+        return structuredClone(cacheEntry.value);
       } else if (cacheEntry instanceof ClientSideCacheEntryPromise) { // 2b2
         // unsure if this should be considered a cache hit, a miss, or neither?
         reply = await cacheEntry.promise;
@@ -171,15 +168,14 @@ export class BasicClientSideCache extends ClientSideCacheProvider {
     // 3b
     if (cacheEntry.validate()) { // revalidating promise entry (dont save value, if promise entry has been invalidated)
       // 3c
-      console.log("saving value to cache");
       cacheEntry = this.createValueEntry(client, val);
       this.set(cacheKey, cacheEntry, parser.keys);
       this.emit("cached-key", cacheKey);
     } else {
-      console.log("cache entry for key got invalidated between execution and saving, so not saving");
+//      console.log("cache entry for key got invalidated between execution and saving, so not saving");
     }
 
-    return val;
+    return structuredClone(val);
   }
 
   override trackingOn() {
@@ -189,21 +185,24 @@ export class BasicClientSideCache extends ClientSideCacheProvider {
   override invalidate(key: RedisArgument | null) {
     if (key === null) {
       this.clear(false);
+      this.emit("invalidate", key);
+
       return;
     }
-    const set = this.#keyToCacheKeySetMap.get(key.toString());
-    if (set) {
-     for (const cacheKey of set) {
-        console.log(`invalidate: got ${cacheKey} from key ${key.toString()} set`);
+
+    const keySet = this.#keyToCacheKeySetMap.get(key.toString());
+    if (keySet) {     
+      for (const cacheKey of keySet) {
         const entry = this.#cacheKeyToEntryMap.get(cacheKey);
         if (entry) {
           entry.invalidate();
         }
         this.#cacheKeyToEntryMap.delete(cacheKey);
-        this.emit("invalidate", cacheKey);
       }
       this.#keyToCacheKeySetMap.delete(key.toString());
     }
+
+    this.emit('invalidate', key);
   }
 
   override clear(reset = true) {
@@ -305,9 +304,9 @@ export class BasicClientSideCache extends ClientSideCacheProvider {
     this.clear();
   }
 
-  override onClose() { }
-    
-  override onDestroy() { }
+  override onClose() {
+    this.clear();
+  }
 
   /**
    * @internal
@@ -366,7 +365,11 @@ export abstract class PooledClientSideCacheProvider extends BasicClientSideCache
     return super.has(cacheKey);
   }  
 
-  onConnect(factory: () => CachingClientType) {};
+  onPoolConnect(factory: () => CachingClientType) {};
+
+  onPoolClose() {
+    this.clear();
+  };
 }
 
 // doesn't do anything special in pooling, clears cache on every client disconnect
@@ -382,6 +385,14 @@ export class BasicPooledClientSideCache extends PooledClientSideCacheProvider {
   override removeClient(client: CachingClientType): void {
     return;
   }
+
+  override onError() {
+    this.clear(false);
+  }
+
+  override onClose() {
+    this.clear(false);
+  }
 }
 
 class PooledClientSideCacheEntryValue extends ClientSideCacheEntryValue {
@@ -443,7 +454,9 @@ export class PooledNoRedirectClientSideCache extends BasicPooledClientSideCache
   }
 
   // don't clear cache on error here
-  override onError(): void {}
+  override onError() {}
+
+  override onClose() {}
 }
 
 // Only clears cache on "management"/"redirect" client disconnect
@@ -482,7 +495,7 @@ export class PooledRedirectClientSideCache extends PooledClientSideCacheProvider
 
   override onError(): void {};
 
-  override async onConnect(factory: () => CachingClientType) {
+  override async onPoolConnect(factory: () => CachingClientType) {
     const client = factory();
     this.#redirectClient = client;
 
@@ -502,20 +515,10 @@ export class PooledRedirectClientSideCache extends PooledClientSideCacheProvider
     }
   }
 
-  onDestroy() {
-    this.clear();
+  override onClose() {};
 
-    if (this.#redirectClient) {
-      this.#id = undefined;
-      const client = this.#redirectClient;
-      this.#redirectClient = undefined;
-
-      client.destroy();
-    }
-  }
-
-  override async onClose() {
-    this.clear();
+  override onPoolClose() {
+    super.onPoolClose();
 
     if (this.#redirectClient) {
       this.#id = undefined;

packages/client/lib/client/cache.spec.ts

@@ -113,7 +113,6 @@ export default class RedisCommandsQueue {
         if (!this.#onPush(push)) {
           switch (push[0].toString()) {
             case "invalidate": {
-              console.log("invalidate push message");
               if (this.#invalidateCallback) {
                 if (push[1] !== null) {
                   for (const key of push[1]) {

packages/client/lib/client/cache.ts

@@ -1110,6 +1110,7 @@ export default class RedisClient<
     return new Promise<void>(resolve => {
       clearTimeout(this._self.#pingTimer);
       this._self.#socket.close();
+      this._self.#clientSideCache?.onClose();
 
       if (this._self.#queue.isEmpty()) {
         this._self.#socket.destroySocket();
@@ -1134,6 +1135,7 @@ export default class RedisClient<
     clearTimeout(this._self.#pingTimer);
     this._self.#queue.flushAll(new DisconnectsClientError());
     this._self.#socket.destroy();
+    this._self.#clientSideCache?.onClose();
   }
 
   ref() {

packages/client/lib/client/commands-queue.ts

@@ -313,7 +313,7 @@ export class RedisClientPool<
     this._self.#isOpen = true;
 
     try {
-      this._self.#clientSideCache?.onConnect(this._self.#clientFactory);
+      this._self.#clientSideCache?.onPoolConnect(this._self.#clientFactory);
     } catch (err) {
       this.destroy();
       throw err;
@@ -476,9 +476,9 @@ export class RedisClientPool<
         promises.push(client.close());
       }
 
-      promises.push(this._self.#clientSideCache?.onClose());
- 
       await Promise.all(promises);
+
+      this.#clientSideCache?.onPoolClose();
 
       this._self.#idleClients.reset();
       this._self.#clientsInUse.reset();
@@ -499,7 +499,7 @@ export class RedisClientPool<
       client.destroy();
     }
 
-    this._self.#clientSideCache?.onDestroy();
+    this._self.#clientSideCache?.onPoolClose();
 
     this._self.#clientsInUse.reset();
 

packages/client/lib/client/index.ts

@@ -328,6 +328,7 @@ export default class RedisClusterSlots<
     return this.#clientFactory(
       this.#clientOptionsDefaults({
         clientSideCache: this.clientSideCache,
+        RESP: this.#options.RESP,
         socket: this.#getNodeAddress(node.address) ?? {
           host: node.host,
           port: node.port

packages/client/lib/client/pool.ts

@@ -595,10 +595,12 @@ export default class RedisCluster<
   }
 
   close() {
+    this.#slots.clientSideCache?.onPoolClose();
     return this._self.#slots.close();
   }
 
   destroy() {
+    this.#slots.clientSideCache?.onPoolClose();
     return this._self.#slots.destroy();
   }
 

packages/client/lib/cluster/cluster-slots.ts

@@ -29,12 +29,7 @@ export function parseGeoSearchArguments(
   from: GeoSearchFrom,
   by: GeoSearchBy,
   options?: GeoSearchOptions,
-  store?: RedisArgument
 ) {
-  if (store !== undefined) {
-    parser.pushKey(store);
-  }
-
   parser.pushKey(key);
 
   if (typeof from === 'string' || from instanceof Buffer) {

packages/client/lib/cluster/index.ts

@@ -17,7 +17,12 @@ export default {
     options?: GeoSearchStoreOptions
   ) {
     parser.push('GEOSEARCHSTORE');
-    parseGeoSearchArguments(parser, source, from, by, options, destination);
+
+    if (destination !== undefined) {
+      parser.pushKey(destination);
+    }
+  
+    parseGeoSearchArguments(parser, source, from, by, options);
 
     if (options?.STOREDIST) {
       parser.push('STOREDIST');

packages/client/lib/commands/GEOSEARCH.ts

@@ -15,6 +15,7 @@ import RedisBloomModules from '@redis/bloom';
 import { BasicPooledClientSideCache } from '../client/cache';
 import { RedisTcpSocketOptions } from '../client/socket';
 import { SQUARE_SCRIPT } from '../client/index.spec';
+import { once } from 'node:events';
 
 const execAsync = promisify(exec);
 
@@ -78,7 +79,7 @@ async function steadyState(frame: SentinelFramework) {
 }
 
 ["redis-sentinel-test-password", undefined].forEach(function (password) {
-  describe.skip(`Sentinel - password = ${password}`, () => {
+  describe(`Sentinel - password = ${password}`, () => {
     const config: RedisSentinelConfig = { sentinelName: "test", numberOfNodes: 3, password: password };
     const frame = new SentinelFramework(config);
     let tracer = new Array<string>();
@@ -1137,22 +1138,10 @@ async function steadyState(frame: SentinelFramework) {
         await nodeAddedPromise; 
       })
 
-      it('with client side caching', async function() {
-        let invalidateResolve;
-        const invalidatePromise = new Promise((res, rej) => {
-          invalidateResolve = res;
-        })
-      
+      it('with client side caching', async function() {     
         this.timeout(30000);
         const csc = new BasicPooledClientSideCache();
 
-        csc.on("invalidate", (cacheKey: string) => {
-          if (cacheKey === "GET_x") {
-            console.log(`invalidating ${cacheKey}`);
-            invalidateResolve(true);
-          }
-        })
-
         sentinel = frame.getSentinelClient({nodeClientOptions: {RESP: 3}, clientSideCache: csc, masterPoolSize: 5});
         await sentinel.connect();
 
@@ -1165,6 +1154,7 @@ async function steadyState(frame: SentinelFramework) {
         assert.equal(1, csc.cacheMisses());
         assert.equal(3, csc.cacheHits());
 
+        const invalidatePromise = once(csc, 'invalidate');
         await sentinel.set('x', 2);
         await invalidatePromise;
         await sentinel.get('x');

packages/client/lib/commands/GEOSEARCHSTORE.ts

@@ -273,10 +273,6 @@ export default class RedisSentinel<
 
     this.#options = options;
 
-    if (options.replicaPoolSize != 0 && options.clientSideCache) {
-      throw new Error("cannot use replica reads and client side cache together");
-    }
-
     if (options.commandOptions) {
       this.#commandOptions = options.commandOptions;
     }
@@ -848,9 +844,7 @@ class RedisSentinelInternal<
 
     this.#isReady = false;
 
-    if (this.#clientSideCache) {
-      this.#clientSideCache.onClose();
-    }
+    this.#clientSideCache?.onPoolClose();
 
     if (this.#scanTimer) {
       clearInterval(this.#scanTimer);
@@ -900,6 +894,8 @@ class RedisSentinelInternal<
 
     this.#isReady = false;
 
+    this.#clientSideCache?.onPoolClose();
+
     if (this.#scanTimer) {
       clearInterval(this.#scanTimer);
       this.#scanTimer = undefined;