Pipeline-based script loading and retries

lib/Redis.ts

@@ -72,12 +72,6 @@ class Redis extends Commander {
   private _autoPipelines = new Map();
   private _runningAutoPipelines = new Set();
 
-  // Prepare a cache of scripts and setup a interval which regularly clears it
-  private _addedScriptHashes: { [sha: string]: true } = {};
-  private _addedScriptHashesCleanInterval?: ReturnType<
-    typeof setInterval
-  > | null = null;
-
   constructor(port: number, host: string, options: RedisOptions);
   constructor(path: string, options: RedisOptions);
   constructor(port: number, options: RedisOptions);
@@ -187,13 +181,6 @@ class Redis extends Commander {
     process.nextTick(this.emit.bind(this, status, arg));
   }
 
-  private clearAddedScriptHashesCleanInterval() {
-    if (this._addedScriptHashesCleanInterval) {
-      clearInterval(this._addedScriptHashesCleanInterval);
-      this._addedScriptHashesCleanInterval = null;
-    }
-  }
-
   /**
    * Create a connection to Redis.
    * This method will be invoked automatically when creating a new Redis instance
@@ -213,18 +200,6 @@ class Redis extends Commander {
         return;
       }
 
-      // Make sure only one timer is active at a time
-      this.clearAddedScriptHashesCleanInterval();
-
-      // Scripts need to get reset on reconnect as redis
-      // might have been restarted or some failover happened
-      this._addedScriptHashes = {};
-
-      // Start the script cache cleaning
-      this._addedScriptHashesCleanInterval = setInterval(() => {
-        this._addedScriptHashes = {};
-      }, this.options.maxScriptsCachingTime);
-
       this.connectionEpoch += 1;
       this.setStatus("connecting");
 
@@ -343,8 +318,6 @@ class Redis extends Commander {
    * If you want to wait for the pending replies, use Redis#quit instead.
    */
   disconnect(reconnect = false) {
-    this.clearAddedScriptHashesCleanInterval();
-
     if (!reconnect) {
       this.manuallyClosing = true;
     }
@@ -628,10 +601,6 @@ class Redis extends Commander {
       command.setTimeout(this.options.commandTimeout);
     }
 
-    if (command.name === "quit") {
-      this.clearAddedScriptHashesCleanInterval();
-    }
-
     let writable =
       this.status === "ready" ||
       (!stream &&
@@ -679,7 +648,16 @@ class Redis extends Commander {
           command.args
         );
       }
-      (stream || this.stream).write(command.toWritable());
+
+      if (stream) {
+        if (stream.isPipeline) {
+          stream.write(command.toWritable(stream.destination.redis.stream));
+        } else {
+          stream.write(command.toWritable(stream));
+        }
+      } else {
+        this.stream.write(command.toWritable(this.stream));
+      }
 
       this.commandQueue.push({
         command: command,

lib/cluster/index.ts

@@ -72,8 +72,6 @@ class Cluster extends Commander {
   _groupsBySlot: number[] = Array(16384);
   private _runningAutoPipelines: Set<string> = new Set();
   private _readyDelayedCallbacks: CallbackFunction[] = [];
-  public _addedScriptHashes: { [key: string]: any } = {};
-  public _addedScriptHashesCleanInterval: NodeJS.Timeout;
 
   /**
    * Every time Cluster#connect() is called, this value will be
@@ -153,13 +151,6 @@ class Cluster extends Commander {
     }
   }
 
-  clearAddedScriptHashesCleanInterval() {
-    if (this._addedScriptHashesCleanInterval) {
-      clearInterval(this._addedScriptHashesCleanInterval);
-      this._addedScriptHashesCleanInterval = null;
-    }
-  }
-
   resetNodesRefreshInterval() {
     if (this.slotsTimer) {
       return;
@@ -195,14 +186,6 @@ class Cluster extends Commander {
         return;
       }
 
-      // Make sure only one timer is active at a time
-      this.clearAddedScriptHashesCleanInterval();
-
-      // Start the script cache cleaning
-      this._addedScriptHashesCleanInterval = setInterval(() => {
-        this._addedScriptHashes = {};
-      }, this.options.maxScriptsCachingTime);
-
       const epoch = ++this.connectionEpoch;
       this.setStatus("connecting");
 
@@ -306,8 +289,6 @@ class Cluster extends Commander {
       debug("closed because %s", reason);
     }
 
-    this.clearAddedScriptHashesCleanInterval();
-
     let retryDelay;
     if (
       !this.manuallyClosing &&
@@ -347,8 +328,6 @@ class Cluster extends Commander {
     const status = this.status;
     this.setStatus("disconnecting");
 
-    this.clearAddedScriptHashesCleanInterval();
-
     if (!reconnect) {
       this.manuallyClosing = true;
     }
@@ -379,8 +358,6 @@ class Cluster extends Commander {
     const status = this.status;
     this.setStatus("disconnecting");
 
-    this.clearAddedScriptHashesCleanInterval();
-
     this.manuallyClosing = true;
 
     if (this.reconnectTimeout) {

lib/command.ts

@@ -9,7 +9,7 @@ import {
   convertObjectToArray,
 } from "./utils";
 import { flatten } from "./utils/lodash";
-import { CallbackFunction, ICommand, CommandParameter } from "./types";
+import { CallbackFunction, ICommand, CommandParameter, NetStream } from "./types";
 
 interface ICommandOptions {
   /**
@@ -269,7 +269,7 @@ export default class Command implements ICommand {
    * @see {@link Redis#sendCommand}
    * @public
    */
-  public toWritable(): string | Buffer {
+  public toWritable(socket: NetStream): string | Buffer {
     let bufferMode = false;
     for (const arg of this.args) {
       if (arg instanceof Buffer) {

lib/pipeline.ts

@@ -1,5 +1,4 @@
 import * as calculateSlot from "cluster-key-slot";
-import * as pMap from "p-map";
 import { exists, hasFlag } from "redis-commands";
 import asCallback from "standard-as-callback";
 import { deprecate } from "util";
@@ -334,71 +333,9 @@ Pipeline.prototype.exec = function (
     }
   }
 
-  // Check whether scripts exists
-  const scripts = [];
-  for (let i = 0; i < this._queue.length; ++i) {
-    const item = this._queue[i];
-
-    if (item.name !== "evalsha") {
-      continue;
-    }
-
-    const script = this._shaToScript[item.args[0]];
-
-    if (
-      !script ||
-      this.redis._addedScriptHashes[script.sha] ||
-      scripts.includes(script)
-    ) {
-      continue;
-    }
-
-    scripts.push(script);
-  }
-
   const _this = this;
-  if (!scripts.length) {
-    return execPipeline();
-  }
-
-  // In cluster mode, always load scripts before running the pipeline
-  if (this.isCluster) {
-    pMap(scripts, (script) => _this.redis.script("load", script.lua), {
-      concurrency: 10,
-    })
-      .then(function () {
-        for (let i = 0; i < scripts.length; i++) {
-          _this.redis._addedScriptHashes[scripts[i].sha] = true;
-        }
-      })
-      .then(execPipeline, this.reject);
-    return this.promise;
-  }
+  execPipeline();
 
-  this.redis
-    .script(
-      "exists",
-      scripts.map(({ sha }) => sha)
-    )
-    .then(function (results) {
-      const pending = [];
-      for (let i = 0; i < results.length; ++i) {
-        if (!results[i]) {
-          pending.push(scripts[i]);
-        }
-      }
-      return Promise.all(
-        pending.map(function (script) {
-          return _this.redis.script("load", script.lua);
-        })
-      );
-    })
-    .then(function () {
-      for (let i = 0; i < scripts.length; i++) {
-        _this.redis._addedScriptHashes[scripts[i].sha] = true;
-      }
-    })
-    .then(execPipeline, this.reject);
   return this.promise;
 
   function execPipeline() {
@@ -414,7 +351,10 @@ Pipeline.prototype.exec = function (
       };
     }
     let bufferMode = false;
+
     const stream = {
+      isPipeline: true,
+      destination: _this.isCluster ? node : { redis: _this.redis },
       write: function (writable) {
         if (writable instanceof Buffer) {
           bufferMode = true;
@@ -442,11 +382,8 @@ Pipeline.prototype.exec = function (
           } else {
             sendData = data;
           }
-          if (_this.isCluster) {
-            node.redis.stream.write(sendData);
-          } else {
-            _this.redis.stream.write(sendData);
-          }
+
+          stream.destination.redis.stream.write(sendData);
 
           // Reset writePending for resending
           writePending = _this._queue.length;

lib/redis/event_handler.ts

@@ -170,8 +170,6 @@ export function closeHandler(self) {
       abortTransactionFragments(self.offlineQueue);
     }
 
-    self.clearAddedScriptHashesCleanInterval();
-
     if (self.manuallyClosing) {
       self.manuallyClosing = false;
       debug("skip reconnecting since the connection is manually closed.");

lib/script.ts

@@ -1,18 +1,10 @@
 import { createHash } from "crypto";
 import Command from "./command";
 import asCallback from "standard-as-callback";
-import { CallbackFunction } from "./types";
-
-function isPromise(obj: any): boolean {
-  return (
-    !!obj &&
-    (typeof obj === "object" || typeof obj === "function") &&
-    typeof obj.then === "function"
-  );
-}
-
+import { CallbackFunction, NetStream } from "./types";
 export default class Script {
   private sha: string;
+  private Command;
 
   constructor(
     private lua: string,
@@ -21,6 +13,31 @@ export default class Script {
     private readOnly: boolean = false
   ) {
     this.sha = createHash("sha1").update(lua).digest("hex");
+
+    const sha = this.sha;
+    const socketHasScriptLoaded = new WeakSet();
+    this.Command = class CustomScriptCommand extends Command {
+      public toWritable(socket: NetStream): string | Buffer {
+        const origReject = this.reject;
+        this.reject = (err) => {
+          if (err.toString().indexOf("NOSCRIPT") !== -1) {
+            socketHasScriptLoaded.delete(socket);
+          }
+          this.reject = origReject;
+          this.reject(err);
+        };
+
+        if (!socketHasScriptLoaded.has(socket)) {
+          socketHasScriptLoaded.add(socket);
+          this.name = "eval";
+          this.args[0] = lua;
+        } else if (this.name === "eval") {
+          this.name = "evalsha";
+          this.args[0] = sha;
+        }
+        return super.toWritable(socket);
+      }
+    };
   }
 
   execute(
@@ -39,28 +56,32 @@ export default class Script {
       options.readOnly = true;
     }
 
-    const evalsha = new Command("evalsha", [this.sha].concat(args), options);
+    const evalsha = new this.Command(
+      "evalsha",
+      [this.sha].concat(args),
+      options
+    );
     evalsha.isCustomCommand = true;
 
-    const result = container.sendCommand(evalsha);
-    if (isPromise(result)) {
-      return asCallback(
-        result.catch((err: Error) => {
-          if (err.toString().indexOf("NOSCRIPT") === -1) {
-            throw err;
-          }
-          return container.sendCommand(
-            new Command("eval", [this.lua].concat(args), options)
-          );
-        }),
-        callback
+    evalsha.promise = evalsha.promise.catch((err: Error) => {
+      if (err.toString().indexOf("NOSCRIPT") === -1) {
+        throw err;
+      }
+
+      // Resend the same custom evalsha command that gets transformed to an eval
+      // in case it's not loaded yet on the connectionDo an eval as fallback, redis will hash and load it
+      const resend = new this.Command(
+        "evalsha",
+        [this.sha].concat(args),
+        options
       );
-    }
+      resend.isCustomCommand = true;
 
-    // result is not a Promise--probably returned from a pipeline chain; however,
-    // we still need the callback to fire when the script is evaluated
-    asCallback(evalsha.promise, callback);
+      const client = container.isPipeline ? container.redis : container;
+      return client.sendCommand(resend);
+    });
 
-    return result;
+    asCallback(evalsha.promise, callback);
+    return container.sendCommand(evalsha);
   }
 }