address review comments

Author: sjpotter

packages/client/lib/client/cache.ts

@@ -1,10 +1,9 @@
 import { EventEmitter } from 'stream';
-import RedisClient, { RedisClientType } from '.';
+import RedisClient from '.';
 import { RedisArgument, ReplyUnion, TransformReply, TypeMapping } from '../RESP/types';
 import { BasicCommandParser } from './parser';
 
 type CachingClient = RedisClient<any, any, any, any, any>;
-type CachingClientType = RedisClientType<any, any, any, any, any>;
 type CmdFunc = () => Promise<ReplyUnion>;
 
 export interface ClientSideCacheConfig {
@@ -23,6 +22,17 @@ interface ClientSideCacheEntry {
   validate(): boolean;
 }
 
+function generateCacheKey(redisArgs: ReadonlyArray<RedisArgument>): string {
+  const tmp = new Array(redisArgs.length*2);
+
+  for (let i = 0; i < redisArgs.length; i++) {
+    tmp[i] = redisArgs[i].length;
+    tmp[i+redisArgs.length] = redisArgs[i];
+  }
+
+  return tmp.join('_');
+}
+
 abstract class ClientSideCacheEntryBase implements ClientSideCacheEntry {
   #invalidated = false;
   readonly #expireTime: number;
@@ -125,7 +135,7 @@ export class BasicClientSideCache extends ClientSideCacheProvider {
   ) {
     let reply: ReplyUnion;
 
-    const cacheKey = parser.cacheKey;
+    const cacheKey = generateCacheKey(parser.redisArgs);
 
     // "2"
     let cacheEntry = this.get(cacheKey);
@@ -339,10 +349,6 @@ export class BasicClientSideCache extends ClientSideCacheProvider {
 export abstract class PooledClientSideCacheProvider extends BasicClientSideCache {
   #disabled = false;
 
-  abstract updateRedirect(id: number): void;
-  abstract addClient(client: CachingClientType): void;
-  abstract removeClient(client: CachingClientType): void;
-
   disable() {
     this.#disabled = true;
   }
@@ -367,27 +373,13 @@ export abstract class PooledClientSideCacheProvider extends BasicClientSideCache
     return super.has(cacheKey);
   }  
 
-  onPoolConnect(factory: () => CachingClientType) {};
-
   onPoolClose() {
     this.clear();
   };
 }
 
 // doesn't do anything special in pooling, clears cache on every client disconnect
 export class BasicPooledClientSideCache extends PooledClientSideCacheProvider {
-
-  override updateRedirect(id: number): void {
-    return;
-  }
-
-  override addClient(client: CachingClientType): void {
-    return;
-  }
-  override removeClient(client: CachingClientType): void {
-    return;
-  }
-
   override onError() {
     this.clear(false);
   }
@@ -459,75 +451,4 @@ export class PooledNoRedirectClientSideCache extends BasicPooledClientSideCache
   override onError() {}
 
   override onClose() {}
-}
-
-// Only clears cache on "management"/"redirect" client disconnect
-export class PooledRedirectClientSideCache extends PooledClientSideCacheProvider {
-  #id?: number;
-  #clients: Set<CachingClientType> = new Set();
-  #redirectClient?: CachingClientType;
-
-  constructor(config: ClientSideCacheConfig) {
-    super(config);
-    this.disable();
-  }
-
-  override trackingOn(): string[] {
-    if (this.#id) {
-      return ['CLIENT', 'TRACKING', 'ON', 'REDIRECT', this.#id.toString()];
-    } else {
-      return [];
-    }
-  }
-
-  override updateRedirect(id: number) {
-    this.#id = id;
-    for (const client of this.#clients) {
-      client.sendCommand(this.trackingOn()).catch(() => {});
-    }
-  }
-
-  override addClient(client: CachingClientType) {
-    this.#clients.add(client);
-  }
-
-  override removeClient(client: CachingClientType) {
-    this.#clients.delete(client);
-  }
-
-  override onError(): void {};
-
-  override async onPoolConnect(factory: () => CachingClientType) {
-    const client = factory();
-    this.#redirectClient = client;
-
-    client.on("error", () => {
-      this.disable();
-      this.clear();
-    }).on("ready", async () => {
-      const clientId = await client.withTypeMapping({}).clientId();
-      this.updateRedirect(clientId);
-      this.enable();
-    })
-
-    try {
-      await client.connect();
-    } catch (err) {
-      throw err;
-    }
-  }
-
-  override onClose() {};
-
-  override onPoolClose() {
-    super.onPoolClose();
-
-    if (this.#redirectClient) {
-      this.#id = undefined;
-      const client = this.#redirectClient;
-      this.#redirectClient = undefined;
-
-      return client.close();
-    }
-  }
-}
+}

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

@@ -111,6 +111,7 @@ export default class RedisCommandsQueue {
       onErrorReply: err => this.#onErrorReply(err),
       onPush: push => {
         if (!this.#onPush(push)) {
+          // currently only supporting "invalidate" over RESP3 push messages
           switch (push[0].toString()) {
             case "invalidate": {
               if (this.#invalidateCallback) {

packages/client/lib/client/index.ts

@@ -318,9 +318,13 @@ export default class RedisClient<
   // was in a watch transaction when 
   // a topology change occured
   #dirtyWatch?: string;
-  #watchEpoch?: number; 
+  #watchEpoch?: number;
   #clientSideCache?: ClientSideCacheProvider;
   #credentialsSubscription: Disposable | null = null;
+  get clientSideCache() {
+    return this._self.#clientSideCache;
+  }
+
 
   get options(): RedisClientOptions<M, F, S, RESP> | undefined {
     return this._self.#options;
@@ -342,7 +346,6 @@ export default class RedisClient<
     return this._self.#socket.socketEpoch;
   }
 
-
   get isWatching() {
     return this._self.#watchEpoch !== undefined;
   }
@@ -542,10 +545,7 @@ export default class RedisClient<
     }
 
     if (this.#clientSideCache) {
-      const tracking = this.#clientSideCache.trackingOn();
-      if (tracking) {
-        commands.push(tracking);
-      }
+      commands.push(this.#clientSideCache.trackingOn());
     }
 
     return commands;
@@ -983,7 +983,7 @@ export default class RedisClient<
     }
 
     const chainId = Symbol('Pipeline Chain'),
-      promise = Promise.allSettled(
+      promise = Promise.all(
         commands.map(({ args }) => this._self.#queue.addCommand(args, {
           chainId,
           typeMapping: this._commandOptions?.typeMapping

packages/client/lib/client/parser.ts

@@ -34,8 +34,14 @@ export class BasicCommandParser implements CommandParser {
   }
 
   get cacheKey() {
-    let cacheKey = this.#redisArgs.map((arg) => arg.length).join('_');
-    return cacheKey + '_' + this.#redisArgs.join('_');
+    const tmp = new Array(this.#redisArgs.length*2);
+
+    for (let i = 0; i < this.#redisArgs.length; i++) {
+      tmp[i] = this.#redisArgs[i].length;
+      tmp[i+this.#redisArgs.length] = this.#redisArgs[i];
+    }
+
+    return tmp.join('_');
   }
 
   push(...arg: Array<RedisArgument>) {

packages/client/lib/client/pool.ts

@@ -7,7 +7,7 @@ import { TimeoutError } from '../errors';
 import { attachConfig, functionArgumentsPrefix, getTransformReply, scriptArgumentsPrefix } from '../commander';
 import { CommandOptions } from './commands-queue';
 import RedisClientMultiCommand, { RedisClientMultiCommandType } from './multi-command';
-import { BasicPooledClientSideCache, ClientSideCacheConfig, PooledClientSideCacheProvider, PooledNoRedirectClientSideCache, PooledRedirectClientSideCache } from './cache';
+import { BasicPooledClientSideCache, ClientSideCacheConfig, PooledClientSideCacheProvider } from './cache';
 import { BasicCommandParser } from './parser';
 import SingleEntryCache from '../single-entry-cache';
 
@@ -222,6 +222,9 @@ export class RedisClientPool<
   }
 
   #clientSideCache?: PooledClientSideCacheProvider;
+  get clientSideCache() {
+    return this._self.#clientSideCache;
+  }
 
   /**
    * You are probably looking for {@link RedisClient.createPool `RedisClient.createPool`},
@@ -248,8 +251,7 @@ export class RedisClientPool<
       } else {
         const cscConfig = options.clientSideCache;
         this.#clientSideCache = clientOptions.clientSideCache = new BasicPooledClientSideCache(cscConfig);
-        this.#clientSideCache = clientOptions.clientSideCache = new PooledNoRedirectClientSideCache(cscConfig);
-        this.#clientSideCache = clientOptions.clientSideCache = new PooledRedirectClientSideCache(cscConfig);
+//        this.#clientSideCache = clientOptions.clientSideCache = new PooledNoRedirectClientSideCache(cscConfig);
       }
     }
 
@@ -319,13 +321,6 @@ export class RedisClientPool<
     if (this._self.#isOpen) return; // TODO: throw error?
     this._self.#isOpen = true;
 
-    try {
-      this._self.#clientSideCache?.onPoolConnect(this._self.#clientFactory);
-    } catch (err) {
-      this.destroy();
-      throw err;
-    }
-
     const promises = [];
     while (promises.length < this._self.#options.minimum) {
       promises.push(this._self.#create());
@@ -341,18 +336,14 @@ export class RedisClientPool<
     return this as unknown as RedisClientPoolType<M, F, S, RESP, TYPE_MAPPING>;
   }
 
-  async #create(redirect?: boolean) {
+  async #create() {
     const node = this._self.#clientsInUse.push(
       this._self.#clientFactory()
         .on('error', (err: Error) => this.emit('error', err))
     );
 
     try {
       const client = node.value;
-      if (this._self.#clientSideCache) {
-        this._self.#clientSideCache.addClient(node.value);
-      }
-
       await client.connect();
     } catch (err) {
       this._self.#clientsInUse.remove(node);
@@ -443,7 +434,6 @@ export class RedisClientPool<
     for (let i = 0; i < toDestroy; i++) {
       // TODO: shift vs pop
       const client = this.#idleClients.shift()!
-      this.#clientSideCache?.removeClient(client);
       client.destroy();
     }
   }

packages/client/lib/cluster/index.ts

@@ -277,6 +277,10 @@ export default class RedisCluster<
     return this._self.#slots.slots;
   }
 
+  get clientSideCache() {
+    return this._self.#slots.clientSideCache;
+  }
+
   /**
    * An array of the cluster masters.
    * Use with {@link RedisCluster.prototype.nodeClient} to get the client for a specific master node.

packages/client/lib/sentinel/index.ts

@@ -16,7 +16,7 @@ import { RedisVariadicArgument } from '../commands/generic-transformers';
 import { WaitQueue } from './wait-queue';
 import { TcpNetConnectOpts } from 'node:net';
 import { RedisTcpSocketOptions } from '../client/socket';
-import { BasicPooledClientSideCache, PooledClientSideCacheProvider, PooledNoRedirectClientSideCache, PooledRedirectClientSideCache } from '../client/cache';
+import { BasicPooledClientSideCache, PooledClientSideCacheProvider } from '../client/cache';
 
 interface ClientInfo {
   id: number;
@@ -302,6 +302,10 @@ export default class RedisSentinel<
   #masterClientCount = 0;
   #masterClientInfo?: ClientInfo;
 
+  get clientSideCache() {
+    return this._self.#internal.clientSideCache;
+  }
+
   constructor(options: RedisSentinelOptions<M, F, S, RESP, TYPE_MAPPING>) {
     super();
 
@@ -618,7 +622,7 @@ class RedisSentinelInternal<
 
   readonly #name: string;
   readonly #nodeClientOptions: RedisClientOptions<M, F, S, RESP, TYPE_MAPPING, RedisTcpSocketOptions>;
-  readonly #sentinelClientOptions: RedisClientOptions<typeof RedisSentinelModule, F, S, RESP, TYPE_MAPPING, RedisTcpSocketOptions>;
+  readonly #sentinelClientOptions: RedisClientOptions<typeof RedisSentinelModule, RedisFunctions, RedisScripts, RespVersions, TypeMapping, RedisTcpSocketOptions>;
   readonly #scanInterval: number;
   readonly #passthroughClientErrorEvents: boolean;
 
@@ -679,8 +683,7 @@ class RedisSentinelInternal<
       } else {
         const cscConfig = options.clientSideCache;
         this.#clientSideCache = this.#nodeClientOptions.clientSideCache = new BasicPooledClientSideCache(cscConfig);
-        this.#clientSideCache = this.#nodeClientOptions.clientSideCache = new PooledNoRedirectClientSideCache(cscConfig);
-        this.#clientSideCache = this.#nodeClientOptions.clientSideCache = new PooledRedirectClientSideCache(cscConfig);
+//        this.#clientSideCache = this.#nodeClientOptions.clientSideCache = new PooledNoRedirectClientSideCache(cscConfig);
       }
     }