RedissonRateLimiter.release() causes availablePermits() to return a value greater than the rate. #7329

import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;

public class RedissonConfig {

    public RedissonClient create() {
        Config config = new Config();
        config.useSingleServer().setAddress("redis://127.0.0.1:6379");
        return Redisson.create(config);
    }
}

@Override
public RFuture<Long> availablePermitsAsync() {
    return commandExecutor.evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_LONG,
            "local rate = redis.call('hget', KEYS[1], 'rate');"
          + "local interval = redis.call('hget', KEYS[1], 'interval');"
          + "local type = redis.call('hget', KEYS[1], 'type');"
          + "assert(rate ~= false and interval ~= false and type ~= false, 'RateLimiter is not initialized')"

          + "local valueName = KEYS[2];"
          + "local permitsName = KEYS[4];"
          + "if type == '1' then "
              + "valueName = KEYS[3];"
              + "permitsName = KEYS[5];"
          + "end;"

          + "local currentValue = redis.call('get', valueName); "
          + "if currentValue == false then "
                 + "redis.call('set', valueName, rate); "
                 + "return rate; "
          + "else "
                 + "local expiredValues = redis.call('zrangebyscore', permitsName, 0, tonumber(ARGV[1]) - interval); "
                 + "local released = 0; "
                 + "for i, v in ipairs(expiredValues) do "
                      + "local random, permits = struct.unpack('Bc0I', v);"
                      + "released = released + permits;"
                 + "end; "

                 + "if released > 0 then "
                      + "redis.call('zremrangebyscore', permitsName, 0, tonumber(ARGV[1]) - interval); "
                    // Double-counting: permits already returned by release() are added again here, and unlike release() this has no clamp to rate.
                      + "currentValue = tonumber(currentValue) + released; "
                      + "redis.call('set', valueName, currentValue);"
                 + "end;"

                 + "return currentValue; "
          + "end;",
            Arrays.asList(getRawName(), getValueName(), getClientValueName(), getPermitsName(), getClientPermitsName()),
            System.currentTimeMillis());
}

@Override
public RFuture<Void> releaseAsync(long permits) {
    if (permits < 0) {
        throw new IllegalArgumentException("Permits amount can't be negative");
    }
    if (permits == 0) {
        return CompletableFutureWrapper.completedNull();
    }
    return commandExecutor.evalWriteAsync(getRawName(), StringCodec.INSTANCE, RedisCommands.EVAL_VOID,
            "local rate = redis.call('hget', KEYS[1], 'rate');"
            + "local interval = redis.call('hget', KEYS[1], 'interval');"
            + "local type = redis.call('hget', KEYS[1], 'type');"
            + "assert(rate ~= false and interval ~= false and type ~= false, 'RateLimiter is not initialized');"

            + "local valueName = KEYS[2];"
            + "local permitsName = KEYS[4];"
            + "if type == '1' then "
            + "    valueName = KEYS[3];"
            + "    permitsName = KEYS[5];"
            + "end;"

            + "local currentValue = redis.call('get', valueName);"
            + "if currentValue == false then "
            + "    currentValue = tonumber(rate);"
            + "else "
            + "    currentValue = tonumber(currentValue);"
            + "end;"

            + "local newValue = currentValue + tonumber(ARGV[1]);"
            + "if newValue > tonumber(rate) then "
            + "    newValue = tonumber(rate);"
            + "end;"
            + "redis.call('set', valueName, newValue);"
            // the acquired permit's entry in permitsName is not removed here, so availablePermits() later counts it a second time.

            + "local keepAliveTime = redis.call('hget', KEYS[1], 'keepAliveTime');"
            + "if (keepAliveTime ~= false and tonumber(keepAliveTime) > 0) then "
            + "    redis.call('pexpire', KEYS[1], keepAliveTime);"
            + "    redis.call('pexpire', valueName, keepAliveTime);"
            + "    redis.call('pexpire', permitsName, keepAliveTime);"
            + "else "
            + "    local ttl = redis.call('pttl', KEYS[1]);"
            + "    if ttl > 0 then "
            + "        redis.call('pexpire', valueName, ttl);"
            + "        redis.call('pexpire', permitsName, ttl);"
            + "    end;"
            + "end;",
            Arrays.asList(getRawName(), getValueName(), getClientValueName(), getPermitsName(), getClientPermitsName()), permits);
}

public void run() throws InterruptedException {
    RateLimiterConfig config = new RateLimiterConfig();
    RRateLimiter limiter = config.create();
    System.out.println("init permits = " + limiter.availablePermits());

    long startNano = System.nanoTime();
    long endAt = System.nanoTime() + SimulationConfig.DURATION_SEC * 1_000_000_000L;
    long count = 0;

    List<long[]> samples = new ArrayList<>();

    while (System.nanoTime() < endAt) {
        limiter.tryAcquire(1);
        Thread.sleep(SimulationConfig.RELEASE_DELAY_MS);
        limiter.release(1);
        long permits = limiter.availablePermits();
        long tMs = (System.nanoTime() - startNano) / 1_000_000;
        long history = config.historySize();
        samples.add(new long[]{tMs, permits, history});
        count++;
    }

    writeJson(samples);
}