fix(worker): fix worker lock algorithm

lemonde · Jul 26, 2022 · ccf24ef · ccf24ef
1 parent bd25d89
commit ccf24ef
Showing 1 changed file with 9 additions and 5 deletions.
diff --git a/lib/locky.js b/lib/locky.js
@@ -9,6 +9,7 @@ const resourceKey = require("./resource-key");
 /**
  * @typedef RedisPromises
  * @property {ReturnType<import('util').promisify<RedisClient["get"]>>} get
+ * @property {ReturnType<import('util').promisify<RedisClient["set"]>>} set
  * @property {ReturnType<import('util').promisify<RedisClient["pexpire"]>>} pexpire
  * @property {ReturnType<import('util').promisify<RedisClient["quit"]>>} quit
  * @property {ReturnType<import('util').promisify<RedisClient["smembers"]>>} smembers
@@ -24,6 +25,7 @@ const resourceKey = require("./resource-key");
 const promisifyRedisClient = (client) => {
   const promises = {
     get: promisify(client.get.bind(client)),
+    set: promisify(client.set.bind(client)),
     pexpire: promisify(client.pexpire.bind(client)),
     quit: promisify(client.quit.bind(client)),
     smembers: promisify(client.smembers.bind(client)),
@@ -317,12 +319,14 @@ class Locky extends EventEmitter {
     const run = async () => {
       try {
         await this.runOperation(async () => {
-          const trx = this.expirationWorker.redis.multi();
           const key = `${this.prefix}expirate:worker`;
-          trx.setnx(key, "OK");
-          trx.pexpire(key, /** @type {number} */ (this.ttl));
-          const [locked] = await promisify(trx.exec.bind(trx))();
-
+          const locked = await this.redis.promises.set(
+            key,
+            "OK",
+            "NX",
+            "PX",
+            this.ttl
+          );
           if (locked) {
             await this.expireKeys();
             await this.redis.promises.del(key);