process_tick.lua

local process_tick = function (now, always_publish)

  local compute_capacity = function (maxConcurrent, running, reservoir)
    if maxConcurrent ~= nil and reservoir ~= nil then
      return math.min((maxConcurrent - running), reservoir)
    elseif maxConcurrent ~= nil then
      return maxConcurrent - running
    elseif reservoir ~= nil then
      return reservoir
    else
      return nil
    end
  end

  local settings = redis.call('hmget', settings_key,
    'id',
    'maxConcurrent',
    'running',
    'reservoir',
    'reservoirRefreshInterval',
    'reservoirRefreshAmount',
    'lastReservoirRefresh',
    'reservoirIncreaseInterval',
    'reservoirIncreaseAmount',
    'reservoirIncreaseMaximum',
    'lastReservoirIncrease',
    'capacityPriorityCounter',
    'clientTimeout'
  )
  local id = settings[1]
  local maxConcurrent = tonumber(settings[2])
  local running = tonumber(settings[3])
  local reservoir = tonumber(settings[4])
  local reservoirRefreshInterval = tonumber(settings[5])
  local reservoirRefreshAmount = tonumber(settings[6])
  local lastReservoirRefresh = tonumber(settings[7])
  local reservoirIncreaseInterval = tonumber(settings[8])
  local reservoirIncreaseAmount = tonumber(settings[9])
  local reservoirIncreaseMaximum = tonumber(settings[10])
  local lastReservoirIncrease = tonumber(settings[11])
  local capacityPriorityCounter = tonumber(settings[12])
  local clientTimeout = tonumber(settings[13])

  local initial_capacity = compute_capacity(maxConcurrent, running, reservoir)

  --
  -- Process 'running' changes
  --
  local expired = redis.call('zrangebyscore', job_expirations_key, '-inf', '('..now)

  if #expired > 0 then
    redis.call('zremrangebyscore', job_expirations_key, '-inf', '('..now)

    local flush_batch = function (batch, acc)
      local weights = redis.call('hmget', job_weights_key, unpack(batch))
                      redis.call('hdel',  job_weights_key, unpack(batch))
      local clients = redis.call('hmget', job_clients_key, unpack(batch))
                      redis.call('hdel',  job_clients_key, unpack(batch))

      -- Calculate sum of removed weights
      for i = 1, #weights do
        acc['total'] = acc['total'] + (tonumber(weights[i]) or 0)
      end

      -- Calculate sum of removed weights by client
      local client_weights = {}
      for i = 1, #clients do
        local removed = tonumber(weights[i]) or 0
        if removed > 0 then
          acc['client_weights'][clients[i]] = (acc['client_weights'][clients[i]] or 0) + removed
        end
      end
    end

    local acc = {
      ['total'] = 0,
      ['client_weights'] = {}
    }
    local batch_size = 1000

    -- Compute changes to Zsets and apply changes to Hashes
    for i = 1, #expired, batch_size do
      local batch = {}
      for j = i, math.min(i + batch_size - 1, #expired) do
        table.insert(batch, expired[j])
      end

      flush_batch(batch, acc)
    end

    -- Apply changes to Zsets
    if acc['total'] > 0 then
      redis.call('hincrby', settings_key, 'done', acc['total'])
      running = tonumber(redis.call('hincrby', settings_key, 'running', -acc['total']))
    end

    for client, weight in pairs(acc['client_weights']) do
      redis.call('zincrby', client_running_key, -weight, client)
    end
  end

  --
  -- Process 'reservoir' changes
  --
  local reservoirRefreshActive = reservoirRefreshInterval ~= nil and reservoirRefreshAmount ~= nil
  if reservoirRefreshActive and now >= lastReservoirRefresh + reservoirRefreshInterval then
    reservoir = reservoirRefreshAmount
    redis.call('hmset', settings_key,
      'reservoir', reservoir,
      'lastReservoirRefresh', now
    )
  end

  local reservoirIncreaseActive = reservoirIncreaseInterval ~= nil and reservoirIncreaseAmount ~= nil
  if reservoirIncreaseActive and now >= lastReservoirIncrease + reservoirIncreaseInterval then
    local num_intervals = math.floor((now - lastReservoirIncrease) / reservoirIncreaseInterval)
    local incr = reservoirIncreaseAmount * num_intervals
    if reservoirIncreaseMaximum ~= nil then
      incr = math.min(incr, reservoirIncreaseMaximum - (reservoir or 0))
    end
    if incr > 0 then
      reservoir = (reservoir or 0) + incr
    end
    redis.call('hmset', settings_key,
      'reservoir', reservoir,
      'lastReservoirIncrease', lastReservoirIncrease + (num_intervals * reservoirIncreaseInterval)
    )
  end

  --
  -- Clear unresponsive clients
  --
  local unresponsive = redis.call('zrangebyscore', client_last_seen_key, '-inf', (now - clientTimeout))
  local unresponsive_lookup = {}
  local terminated_clients = {}
  for i = 1, #unresponsive do
    unresponsive_lookup[unresponsive[i]] = true
    if tonumber(redis.call('zscore', client_running_key, unresponsive[i])) == 0 then
      table.insert(terminated_clients, unresponsive[i])
    end
  end
  if #terminated_clients > 0 then
    redis.call('zrem', client_running_key,         unpack(terminated_clients))
    redis.call('hdel', client_num_queued_key,      unpack(terminated_clients))
    redis.call('zrem', client_last_registered_key, unpack(terminated_clients))
    redis.call('zrem', client_last_seen_key,       unpack(terminated_clients))
  end

  --
  -- Broadcast capacity changes
  --
  local final_capacity = compute_capacity(maxConcurrent, running, reservoir)

  if always_publish or (initial_capacity ~= nil and final_capacity == nil) then
    -- always_publish or was not unlimited, now unlimited
    redis.call('publish', 'b_'..id, 'capacity:'..(final_capacity or ''))

  elseif initial_capacity ~= nil and final_capacity ~= nil and final_capacity > initial_capacity then
    -- capacity was increased
    -- send the capacity message to the limiter having the lowest number of running jobs
    -- the tiebreaker is the limiter having not registered a job in the longest time

    local lowest_concurrency_value = nil
    local lowest_concurrency_clients = {}
    local lowest_concurrency_last_registered = {}
    local client_concurrencies = redis.call('zrange', client_running_key, 0, -1, 'withscores')

    for i = 1, #client_concurrencies, 2 do
      local client = client_concurrencies[i]
      local concurrency = tonumber(client_concurrencies[i+1])

      if (
        lowest_concurrency_value == nil or lowest_concurrency_value == concurrency
      ) and (
        not unresponsive_lookup[client]
      ) and (
        tonumber(redis.call('hget', client_num_queued_key, client)) > 0
      ) then
        lowest_concurrency_value = concurrency
        table.insert(lowest_concurrency_clients, client)
        local last_registered = tonumber(redis.call('zscore', client_last_registered_key, client))
        table.insert(lowest_concurrency_last_registered, last_registered)
      end
    end

    if #lowest_concurrency_clients > 0 then
      local position = 1
      local earliest = lowest_concurrency_last_registered[1]

      for i,v in ipairs(lowest_concurrency_last_registered) do
        if v < earliest then
          position = i
          earliest = v
        end
      end

      local next_client = lowest_concurrency_clients[position]
      redis.call('publish', 'b_'..id,
        'capacity-priority:'..(final_capacity or '')..
        ':'..next_client..
        ':'..capacityPriorityCounter
      )
      redis.call('hincrby', settings_key, 'capacityPriorityCounter', '1')
    else
      redis.call('publish', 'b_'..id, 'capacity:'..(final_capacity or ''))
    end
  end

  return {
    ['capacity'] = final_capacity,
    ['running'] = running,
    ['reservoir'] = reservoir
  }
end