Add expiration to lock queue, remove token from queue on timeout (#62)

Fixes #61.
amphp · Oct 24, 2019 · fe670a2 · fe670a2
1 parent 494a32c
commit fe670a2
Showing 1 changed file with 13 additions and 1 deletion.
diff --git a/src/Mutex/Mutex.php b/src/Mutex/Mutex.php
@@ -27,6 +27,7 @@ final class Mutex implements KeyedMutex
 
 local token = ARGV[1]
 local ttl = ARGV[2]
+local queueTtl = ARGV[3]
 
 if redis.call("exists", lock) == 0 then
     if redis.call("llen", queue) == 0 then
@@ -54,6 +55,8 @@ final class Mutex implements KeyedMutex
         if queued == token then
             return 2
         else
+            redis.call("pexpire", queue, queueTtl)
+
             return -1 - position
         end
     end
@@ -66,11 +69,14 @@ final class Mutex implements KeyedMutex
     local queued_tokens = redis.call("lrange", queue, 0, -1)
     for i=1,#queued_tokens do
         if queued_tokens[i] == token then
+            redis.call("pexpire", queue, queueTtl)
+
             return -1 - i
         end
     end
 
     redis.call("rpush", queue, token)
+    redis.call("pexpire", queue, queueTtl)
 
     return -1 - redis.call("llen", queue)
 end
@@ -161,11 +167,17 @@ public function acquire(string $key): Promise
                 $result = yield $this->sharedConnection->eval(
                     self::LOCK,
                     ["{$prefix}lock:{$key}", "{$prefix}lock-queue:{$key}"],
-                    [$token, $this->options->getLockExpiration()]
+                    [$token, $this->options->getLockExpiration(), $this->options->getLockExpiration() + $this->options->getLockTimeout()]
                 );
 
                 if ($result < 1) {
                     if ($attempts > 2 && \microtime(true) * 1000 > $timeLimit) {
+                        // In very rare cases we might not get the lock, but are at the head of the queue and another
+                        // client moves us into the lock position. Deleting the token from the queue and afterwards
+                        // unlocking solves this. No yield required, because we use the same connection.
+                        $this->sharedConnection->getList("{$prefix}lock-queue:{$key}")->remove($token);
+                        $this->unlock("{$prefix}lock:{$key}", $token);
+
                         throw new LockException('Failed to acquire lock for ' . $key . ' within ' . $this->options->getLockTimeout() . ' ms');
                     }