for "redirect" client, don't treat it as part of pool

this enables a lot of cleanup

Author: sjpotter

packages/client/lib/client/cache.ts

@@ -87,6 +87,7 @@ export abstract class ClientSideCacheProvider {
   abstract clear(): void;
   abstract cacheHits(): number;
   abstract cacheMisses(): number;
+  abstract clearOnReconnect(): boolean;
 }
 
 export class BasicClientSideCache extends ClientSideCacheProvider {
@@ -108,6 +109,10 @@ export class BasicClientSideCache extends ClientSideCacheProvider {
     this.#lru = lru;
   }
 
+  override clearOnReconnect(): boolean {
+    return true;
+  }
+
   /* logic of how caching works:
 
   1. every command that's cachable provide a getCacheInfo(args) function
@@ -399,17 +404,17 @@ export class PooledNoRedirectClientSideCache extends BasicPooledClientSideCache
     return new PooledClientSideCacheEntryPromise(this.ttl, creator, sendCommandPromise);
   }
 
-  // doesn't provide any mechanism for clearing cache
-  override clear(): void {
-    return;
+  override clearOnReconnect(): boolean {
+    return false;
   }
 }
 
 // Only clears cache on "management"/"redirect" client disconnect
 export class PooledRedirectClientSideCache extends PooledClientSideCacheProvider {
   #id?: number;
-  #clients: Set<RedisClient<any, any, any, any, any>> = new Set();
+  #clients: Set<CachingClient> = new Set();
   #disabled = true;
+  #redirectClient?: CachingClient;
 
   constructor(ttl: number, maxEntries: number, lru: boolean) {
     super(ttl, maxEntries, lru);
@@ -450,7 +455,7 @@ export class PooledRedirectClientSideCache extends PooledClientSideCacheProvider
   override updateRedirect(id: number) {
     this.#id = id;
     for (const client of this.#clients) {
-      client.sendCommand(this.trackingOn())
+      client.sendCommand(this.trackingOn()).catch(() => {});
     }
   }
 
@@ -462,13 +467,27 @@ export class PooledRedirectClientSideCache extends PooledClientSideCacheProvider
     this.#clients.delete(client);
   }
 
-  // only clear via redirect client, as this will be on the regular client emits
-  override clear() {
-    return;
+  override clearOnReconnect(): boolean {
+    return false;
+  }
+
+  setRedirectClient(client: CachingClient) {
+    this.#redirectClient = client;
   }
 
-  // this will be attached to the redirect client emits.
-  clearPool() {
-    super.clear();
+  destroy() {
+    this.clear();
+
+    if (this.#redirectClient) {
+      this.#redirectClient.destroy();
+    }
+  }
+
+  async close() {
+    this.clear();
+
+    if (this.#redirectClient) {
+      return this.#redirectClient.close();
+    }
   }
 }

packages/client/lib/client/index.ts

@@ -316,9 +316,9 @@ export default class RedisClient<
   #monitorCallback?: MonitorCallback<TYPE_MAPPING>;
   private _self = this;
   private _commandOptions?: CommandOptions<TYPE_MAPPING>;
-  #annotations = new Map<string, string>();
-
+  
   #clientSideCache?: ClientSideCacheProvider;
+  #clearOnReconnect = false;
 
   get options(): RedisClientOptions<M, F, S, RESP> | undefined {
     return this._self.#options;
@@ -347,11 +347,14 @@ export default class RedisClient<
     this.#queue = this.#initiateQueue();
     this.#socket = this.#initiateSocket();
 
-    if (options?.clientSideCache instanceof ClientSideCacheProvider) {
-      this.#clientSideCache = options.clientSideCache;
-    } else if (options?.clientSideCache !== undefined) {
-      const cscConfig = options.clientSideCache;
-      this.#clientSideCache = new BasicClientSideCache(cscConfig.ttl, cscConfig.maxEntries, cscConfig.lru);
+    if (options?.clientSideCache) {
+      if (options.clientSideCache instanceof ClientSideCacheProvider) {
+        this.#clientSideCache = options.clientSideCache;
+      } else {
+        const cscConfig = options.clientSideCache;
+        this.#clientSideCache = new BasicClientSideCache(cscConfig.ttl, cscConfig.maxEntries, cscConfig.lru);
+      }
+      this.#clearOnReconnect = this.#clientSideCache.clearOnReconnect();
     }
   }
 
@@ -483,7 +486,9 @@ export default class RedisClient<
       })
       .on('error', err => {
         this.emit('error', err);
-        this.#clientSideCache?.clear(); // clear at failure time
+        if (this.#clearOnReconnect) {
+          this.#clientSideCache!.clear();
+        }
         if (this.#socket.isOpen && !this.#options?.disableOfflineQueue) {
           this.#queue.flushWaitingForReply(err);
         } else {
@@ -641,26 +646,6 @@ export default class RedisClient<
     }
   }
 
-  setAnnotation(key: string, val: string) {
-    this._self.#annotations.set(key, val);
-  }
-
-  getAnnotation(key: string) {
-    return this._self.#annotations.get(key);
-  }
-
-  clearAnnotations() {
-    return this._self.#annotations.clear();
-  }
-
-  hasAnnotation(key: string) {
-    return this._self.#annotations.has(key);
-  }
-
-  deleteAnnotation(key: string) {
-    return this._self.#annotations.delete(key);
-  }
-
   async SELECT(db: number): Promise<void> {
     await this.sendCommand(['SELECT', db.toString()]);
     this._self.#selectedDB = db;

packages/client/lib/client/pool.ts

@@ -311,22 +311,29 @@ export class RedisClientPool<
     if (this._self.#isOpen) return; // TODO: throw error?
     this._self.#isOpen = true;
 
-    let count = 0;
     if (this._self.#clientSideCache instanceof PooledRedirectClientSideCache) {
-      if (this._self.#options.minimum < 1) {
-        this._self.#options.minimum = 1;
-      }
-      count = 1;
+      const client = this._self.#clientFactory();
+      const cache = this._self.#clientSideCache;
+      client.on("error", () => {
+        cache.disable();
+        cache.clear();
+      }).on("ready", async () => {
+        const clientId = await client.clientId() as number;
+        cache.updateRedirect(clientId);
+        cache.enable();
+      })
+
       try {
-        await this._self.#create(true)
+        await client.connect();
+        cache.setRedirectClient(client);
       } catch (err) {
         this.destroy();
         throw err;
       }
     }
 
     const promises = [];
-    while (promises.length + count < this._self.#options.minimum) {
+    while (promises.length < this._self.#options.minimum) {
       promises.push(this._self.#create());
     }
 
@@ -348,18 +355,7 @@ export class RedisClientPool<
 
     try {
       const client = node.value;
-      const cache = this._self.#clientSideCache;
-      if (redirect && cache instanceof PooledRedirectClientSideCache) {
-        client.setAnnotation("pool-master", "true");
-        client.on("error", () => {
-          cache.disable();
-        }).on("ready", async () => {
-          const clientId = await client.clientId() as number;
-          cache.updateRedirect(clientId);
-          cache.clearPool();
-          cache.enable();
-        })
-      } else if (this._self.#clientSideCache) {
+      if (this._self.#clientSideCache) {
         this._self.#clientSideCache.addClient(node.value);
       }
 
@@ -453,12 +449,6 @@ export class RedisClientPool<
     for (let i = 0; i < toDestroy; i++) {
       // TODO: shift vs pop
       const client = this.#idleClients.shift()!
-      if (client.hasAnnotation("pool-master")) {
-        console.log("wanted to remove \"master\"/\"redirect\" client");
-        this._self.#idleClients.push(client);
-        continue;
-      }
-
       this.#clientSideCache?.removeClient(client);
       client.destroy();
     }
@@ -505,6 +495,10 @@ export class RedisClientPool<
       for (const client of this._self.#clientsInUse) {
         promises.push(client.close());
       }
+
+      if (this.#clientSideCache instanceof PooledRedirectClientSideCache) {
+        promises.push(this.#clientSideCache.close());
+      }
 
       await Promise.all(promises);
 
@@ -526,6 +520,11 @@ export class RedisClientPool<
     for (const client of this._self.#clientsInUse) {
       client.destroy();
     }
+
+    if (this.#clientSideCache instanceof PooledRedirectClientSideCache) {
+      this.#clientSideCache.destroy();
+    }
+
     this._self.#clientsInUse.reset();
 
     this._self.#isOpen = false;