Skip to content

Commit 91a658e

Browse files
manish-plivomanish-plivo
authored
Merge pull request #18 from ppai-plivo/sharq-requeue-take2
Handle known races caused by requeue script
2 parents 01303e4 + 4a5077d commit 91a658e

File tree

10 files changed

+123
-76
lines changed

10 files changed

+123
-76
lines changed

‎CHANGELOG‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,6 @@
1+
MINOR RELEASE v1.1.0
2+
- Handle known races in requeue logic.
3+
14
MAJOR RELEASE v1.0.0
25
- Upgrade python version to 3.6
36

‎requirements.txt‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
11
argparse==1.2.1
22
msgpack==0.5.6
33
wsgiref==0.1.2
4-
redis-py-cluster==2.0.0
4+
redis-py-cluster==2.1.0

‎setup.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44

55
setup(
66
name='SharQ',
7-
version='1.0.0',
7+
version='1.1.0',
88
url='https://github.com/plivo/sharq',
99
author='Plivo Team',
1010
author_email='voice-team@plivo.com',

‎sharq/queue.py‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -75,6 +75,9 @@ def _load_config(self):
7575
self._config = configparser.SafeConfigParser()
7676
self._config.read(self.config_path)
7777

78+
def redis_client(self):
79+
return self._r
80+
7881
def reload_config(self, config_path=None):
7982
"""Reload the configuration from the new config file if provided
8083
else reload the current config file.
@@ -214,6 +217,13 @@ def dequeue(self, queue_type='default'):
214217
return response
215218

216219
queue_id, job_id, payload, requeues_remaining = dequeue_response
220+
221+
if payload is None:
222+
response = {
223+
'status': 'failure'
224+
}
225+
return response
226+
217227
payload = deserialize_payload(payload)
218228

219229
response = {
@@ -315,6 +325,9 @@ def requeue(self):
315325
'%s:active:queue_type' % self._key_prefix)
316326
for queue_type in active_queue_type_list:
317327
# requeue all expired jobs in all queue types.
328+
329+
queue_type = queue_type.decode('utf-8')
330+
318331
keys = [
319332
self._key_prefix,
320333
queue_type

‎sharq/scripts/lua/dequeue.lua‎

Lines changed: 34 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -10,63 +10,73 @@
1010
-- { queue_id, job_id, payload, requeues_remaining }
1111

1212

13-
local ready_queue_id_list = redis.call('ZRANGEBYSCORE', KEYS[1] .. ':' .. KEYS[2], 0, ARGV[1])
13+
local prefix = KEYS[1]
14+
local queue_type = KEYS[2]
15+
16+
local current_timestamp = ARGV[1]
17+
local job_expiry_interval = ARGV[2]
18+
19+
20+
local ready_queue_id_list = redis.call('ZRANGEBYSCORE', prefix .. ':' .. queue_type, 0, current_timestamp)
1421
if next(ready_queue_id_list) ~= nil then
1522
-- there is a queue ready to be dequeued.
1623
local ready_queue_id = ready_queue_id_list[1]
1724
-- dequeue a job from the job queue.
18-
local job_id = redis.call('LPOP', KEYS[1] .. ':' .. KEYS[2] .. ':' .. ready_queue_id)
25+
local job_id = redis.call('LPOP', prefix .. ':' .. queue_type .. ':' .. ready_queue_id)
1926
-- get the payload for this job
20-
local payload = redis.call('HGET', KEYS[1] .. ':payload', KEYS[2] .. ':' .. ready_queue_id .. ':' .. job_id)
27+
local payload = redis.call('HGET', prefix .. ':payload', queue_type .. ':' .. ready_queue_id .. ':' .. job_id)
2128
-- update the time keeper with the current dequeue time.
22-
local interval = redis.call('HGET', KEYS[1] .. ':interval', KEYS[2] .. ':' .. ready_queue_id)
23-
redis.call('PSETEX', KEYS[1] .. ':' .. KEYS[2] .. ':' .. ready_queue_id .. ':time', ARGV[2], ARGV[1])
29+
redis.call('PSETEX', prefix .. ':' .. queue_type .. ':' .. ready_queue_id .. ':time', job_expiry_interval, current_timestamp)
2430
-- check if there are any more jobs of this queue in the job queue.
25-
if redis.call('LLEN', KEYS[1] .. ':' .. KEYS[2] .. ':' .. ready_queue_id) == 0 then
31+
if redis.call('LLEN', prefix .. ':' .. queue_type .. ':' .. ready_queue_id) == 0 then
2632
-- there are no more jobs of this queue. remove this queue from the ready sorted set.
27-
redis.call('ZREM', KEYS[1] .. ':' .. KEYS[2], ready_queue_id)
33+
redis.call('ZREM', prefix .. ':' .. queue_type, ready_queue_id)
2834
-- now check if the ready sorted set is empty.
29-
if redis.call('EXISTS', KEYS[1] .. ':' .. KEYS[2]) ~= 1 then
35+
if redis.call('EXISTS', prefix .. ':' .. queue_type) ~= 1 then
3036
-- the ready sorted set is empty. remove this 'queue_type' from
3137
-- the metris ready queue type set
32-
redis.call('SREM', KEYS[1] .. ':ready:queue_type', KEYS[2])
38+
redis.call('SREM', prefix .. ':ready:queue_type', queue_type)
3339
end
3440
else
3541
-- there are more jobs in the queue. update the next
3642
-- dequeue time for this queue in the ready sorted set.
37-
local next_dequeue_time = ARGV[1] + interval
38-
redis.call('ZADD', KEYS[1] .. ':' .. KEYS[2], next_dequeue_time, ready_queue_id)
43+
local next_dequeue_time = current_timestamp
44+
local interval = tonumber(redis.call('HGET', prefix .. ':interval', queue_type .. ':' .. ready_queue_id))
45+
if interval then
46+
next_dequeue_time = current_timestamp + interval
47+
end
48+
redis.call('ZADD', prefix .. ':' .. queue_type, next_dequeue_time, ready_queue_id)
3949
end
40-
local job_expiry_time = ARGV[1] + ARGV[2]
50+
local job_expiry_time = current_timestamp + job_expiry_interval
4151
-- finally, add the job_id and queue_id that was dequeued into the active sorted set.
42-
redis.call('ZADD', KEYS[1] .. ':' .. KEYS[2] .. ':active', job_expiry_time, ready_queue_id .. ':' .. job_id)
52+
redis.call('ZADD', prefix .. ':' .. queue_type .. ':active', job_expiry_time, ready_queue_id .. ':' .. job_id)
4353
-- add the queue_type to metrics active queue type set.
44-
redis.call('SADD', KEYS[1] .. ':active:queue_type', KEYS[2])
54+
redis.call('SADD', prefix .. ':active:queue_type', queue_type)
4555

4656
-- get the requeues_remaining for this job
47-
local requeues_remaining = redis.call('HGET', KEYS[1] .. ':' .. KEYS[2] .. ':' .. ready_queue_id .. ':requeues_remaining', job_id)
57+
local requeues_remaining = redis.call('HGET', prefix .. ':' .. queue_type .. ':' .. ready_queue_id .. ':requeues_remaining', job_id)
4858

4959
-- update the metrics counters
5060
-- update global counter.
51-
local timestamp_minute = math.floor(ARGV[1]/60000) * 60000 -- get the epoch for the minute
61+
local timestamp_minute = math.floor(current_timestamp/60000) * 60000 -- get the epoch for the minute
5262
local expiry_time = math.floor((timestamp_minute + 600000) / 1000) -- store the data for 10 minutes.
53-
if redis.call('EXISTS', KEYS[1] .. ':dequeue_counter:' .. timestamp_minute) ~= 1 then
63+
if redis.call('EXISTS', prefix .. ':dequeue_counter:' .. timestamp_minute) ~= 1 then
5464
-- counter does not exists. set the initial value and expiry.
55-
redis.call('SET', KEYS[1] .. ':dequeue_counter:' .. timestamp_minute, 1)
56-
redis.call('EXPIREAT', KEYS[1] .. ':dequeue_counter:' .. timestamp_minute, expiry_time)
65+
redis.call('SET', prefix .. ':dequeue_counter:' .. timestamp_minute, 1)
66+
redis.call('EXPIREAT', prefix .. ':dequeue_counter:' .. timestamp_minute, expiry_time)
5767
else
5868
-- counter already exists. just increment the value.
59-
redis.call('INCR', KEYS[1] .. ':dequeue_counter:' .. timestamp_minute)
69+
redis.call('INCR', prefix .. ':dequeue_counter:' .. timestamp_minute)
6070
end
6171

6272
-- update the current queue counter.
63-
if redis.call('EXISTS', KEYS[1] .. ':' .. KEYS[2] .. ':' .. ready_queue_id .. ':dequeue_counter:' .. timestamp_minute) ~= 1 then
73+
if redis.call('EXISTS', prefix .. ':' .. queue_type .. ':' .. ready_queue_id .. ':dequeue_counter:' .. timestamp_minute) ~= 1 then
6474
-- counter does not exists. set the initial value and expiry.
65-
redis.call('SET', KEYS[1] .. ':' .. KEYS[2] .. ':' .. ready_queue_id .. ':dequeue_counter:' .. timestamp_minute, 1)
66-
redis.call('EXPIREAT', KEYS[1] .. ':' .. KEYS[2] .. ':' .. ready_queue_id .. ':dequeue_counter:' .. timestamp_minute, expiry_time)
75+
redis.call('SET', prefix .. ':' .. queue_type .. ':' .. ready_queue_id .. ':dequeue_counter:' .. timestamp_minute, 1)
76+
redis.call('EXPIREAT', prefix .. ':' .. queue_type .. ':' .. ready_queue_id .. ':dequeue_counter:' .. timestamp_minute, expiry_time)
6777
else
6878
-- counter already exists. just increment the value.
69-
redis.call('INCR', KEYS[1] .. ':' .. KEYS[2] .. ':' .. ready_queue_id .. ':dequeue_counter:' .. timestamp_minute)
79+
redis.call('INCR', prefix .. ':' .. queue_type .. ':' .. ready_queue_id .. ':dequeue_counter:' .. timestamp_minute)
7080
end
7181

7282
return { ready_queue_id, job_id, payload, requeues_remaining }

‎sharq/scripts/lua/enqueue.lua‎

Lines changed: 28 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -13,55 +13,63 @@
1313
-- output:
1414
-- nil
1515

16+
local prefix = KEYS[1]
17+
local queue_type = KEYS[2]
18+
19+
local current_timestamp = ARGV[1]
20+
local queue_id = ARGV[2]
21+
local job_id = ARGV[3]
22+
local payload = ARGV[4]
23+
local interval = ARGV[5]
24+
local requeue_limit = ARGV[6]
1625

1726
-- push the job id into the job queue.
18-
redis.call('RPUSH', KEYS[1] .. ':' .. KEYS[2] .. ':' .. ARGV[2], ARGV[3])
27+
redis.call('RPUSH', prefix .. ':' .. queue_type .. ':' .. queue_id, job_id)
1928

2029
-- update the payload map.
21-
redis.call('HSET', KEYS[1] .. ':payload', KEYS[2] .. ':' .. ARGV[2] .. ':' .. ARGV[3], ARGV[4])
30+
redis.call('HSET', prefix .. ':payload', queue_type .. ':' .. queue_id .. ':' .. job_id, payload)
2231

2332
-- update the interval map.
24-
redis.call('HSET', KEYS[1] .. ':interval', KEYS[2] .. ':' .. ARGV[2], ARGV[5])
33+
redis.call('HSET', prefix .. ':interval', queue_type .. ':' .. queue_id, interval)
2534

2635
-- update the requeue limit map.
27-
redis.call('HSET', KEYS[1] .. ':' .. KEYS[2] .. ':' .. ARGV[2] .. ':requeues_remaining', ARGV[3], ARGV[6])
36+
redis.call('HSET', prefix .. ':' .. queue_type .. ':' .. queue_id .. ':requeues_remaining', job_id, requeue_limit)
2837

2938
-- check if the queue of this job is already present in the ready sorted set.
30-
if not redis.call('ZRANK', KEYS[1] .. ':' .. KEYS[2], ARGV[2]) then
39+
if not redis.call('ZRANK', prefix .. ':' .. queue_type, queue_id) then
3140
-- the ready sorted set is empty, update it and add it to metrics ready queue type set.
32-
redis.call('SADD', KEYS[1] .. ':ready:queue_type', KEYS[2])
33-
if redis.call('EXISTS', KEYS[1] .. ':' .. KEYS[2] .. ':' .. ARGV[2] .. ':time') ~= 1 then
41+
redis.call('SADD', prefix .. ':ready:queue_type', queue_type)
42+
if redis.call('EXISTS', prefix .. ':' .. queue_type .. ':' .. queue_id .. ':time') ~= 1 then
3443
-- time keeper does not exist
3544
-- update the ready sorted set with current time as ready time.
36-
redis.call('ZADD', KEYS[1] .. ':' .. KEYS[2], ARGV[1], ARGV[2])
45+
redis.call('ZADD', prefix .. ':' .. queue_type, current_timestamp, queue_id)
3746
else
3847
-- time keeper exists
39-
local interval = ARGV[5]
40-
local last_dequeue_time = redis.call('GET', KEYS[1] .. ':' .. KEYS[2] .. ':' .. ARGV[2] .. ':time')
48+
local last_dequeue_time = redis.call('GET', prefix .. ':' .. queue_type .. ':' .. queue_id .. ':time')
4149
local ready_time = interval + last_dequeue_time
42-
redis.call('ZADD', KEYS[1] .. ':' .. KEYS[2], ready_time, ARGV[2])
50+
redis.call('ZADD', prefix .. ':' .. queue_type, ready_time, queue_id)
4351
end
4452
end
4553

4654
-- update the metrics counters
4755
-- update global counter.
48-
local timestamp_minute = math.floor(ARGV[1]/60000) * 60000 -- get the epoch for the minute
56+
local timestamp_minute = math.floor(current_timestamp/60000) * 60000 -- get the epoch for the minute
4957
local expiry_time = math.floor((timestamp_minute + 600000) / 1000) -- store the data for 10 minutes.
50-
if redis.call('EXISTS', KEYS[1] .. ':enqueue_counter:' .. timestamp_minute) ~= 1 then
58+
if redis.call('EXISTS', prefix .. ':enqueue_counter:' .. timestamp_minute) ~= 1 then
5159
-- counter does not exists. set the initial value and expiry.
52-
redis.call('SET', KEYS[1] .. ':enqueue_counter:' .. timestamp_minute, 1)
53-
redis.call('EXPIREAT', KEYS[1] .. ':enqueue_counter:' .. timestamp_minute, expiry_time)
60+
redis.call('SET', prefix .. ':enqueue_counter:' .. timestamp_minute, 1)
61+
redis.call('EXPIREAT', prefix .. ':enqueue_counter:' .. timestamp_minute, expiry_time)
5462
else
5563
-- counter already exists. just increment the value.
56-
redis.call('INCR', KEYS[1] .. ':enqueue_counter:' .. timestamp_minute)
64+
redis.call('INCR', prefix .. ':enqueue_counter:' .. timestamp_minute)
5765
end
5866

5967
-- update the current queue counter.
60-
if redis.call('EXISTS', KEYS[1] .. ':' .. KEYS[2] .. ':' .. ARGV[2] .. ':enqueue_counter:' .. timestamp_minute) ~= 1 then
68+
if redis.call('EXISTS', prefix .. ':' .. queue_type .. ':' .. queue_id .. ':enqueue_counter:' .. timestamp_minute) ~= 1 then
6169
-- counter does not exists. set the initial value and expiry.
62-
redis.call('SET', KEYS[1] .. ':' .. KEYS[2] .. ':' .. ARGV[2] .. ':enqueue_counter:' .. timestamp_minute, 1)
63-
redis.call('EXPIREAT', KEYS[1] .. ':' .. KEYS[2] .. ':' .. ARGV[2] .. ':enqueue_counter:' .. timestamp_minute, expiry_time)
70+
redis.call('SET', prefix .. ':' .. queue_type .. ':' .. queue_id .. ':enqueue_counter:' .. timestamp_minute, 1)
71+
redis.call('EXPIREAT', prefix .. ':' .. queue_type .. ':' .. queue_id .. ':enqueue_counter:' .. timestamp_minute, expiry_time)
6472
else
6573
-- counter already exists. just increment the value.
66-
redis.call('INCR', KEYS[1] .. ':' .. KEYS[2] .. ':' .. ARGV[2] .. ':enqueue_counter:' .. timestamp_minute)
74+
redis.call('INCR', prefix .. ':' .. queue_type .. ':' .. queue_id .. ':enqueue_counter:' .. timestamp_minute)
6775
end

‎sharq/scripts/lua/finish.lua‎

Lines changed: 12 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -9,29 +9,34 @@
99
-- output:
1010
-- nil
1111

12+
local prefix = KEYS[1]
13+
local queue_type = KEYS[2]
14+
local queue_id = ARGV[1]
15+
local job_id = ARGV[2]
16+
1217

1318
-- remove the job from active sorted set.
14-
local response = redis.call('ZREM', KEYS[1] .. ':' .. KEYS[2] .. ':active', ARGV[1] .. ':' .. ARGV[2])
19+
local response = redis.call('ZREM', prefix .. ':' .. queue_type .. ':active', queue_id .. ':' .. job_id)
1520
if response ~= 1 then
1621
-- the job was not found in the active sorted set. Non existent job or
1722
-- the job is expired and was requeued back.
1823
return 0
1924
end
2025

2126
-- check if the just-removed job was the last job in the active sorted set.
22-
if redis.call('EXISTS', KEYS[1] .. ':' .. KEYS[2] .. ':active') ~= 1 then
27+
if redis.call('EXISTS', prefix .. ':' .. queue_type .. ':active') ~= 1 then
2328
-- yes. this was the last job. remove this queue_type
2429
-- from the metrics active queue type set.
25-
redis.call('SREM', KEYS[1] .. ':active:queue_type', KEYS[2])
30+
redis.call('SREM', prefix .. ':active:queue_type', queue_type)
2631
end
2732
-- delete the payload related to this job from the payload map.
28-
redis.call('HDEL', KEYS[1] .. ':payload', KEYS[2] .. ':' .. ARGV[1] .. ':' .. ARGV[2])
29-
if redis.call('EXISTS', KEYS[1] .. ':' .. KEYS[2] .. ':' .. ARGV[1]) ~= 1 then
33+
redis.call('HDEL', prefix .. ':payload', queue_type .. ':' .. queue_id .. ':' .. job_id)
34+
if redis.call('EXISTS', prefix .. ':' .. queue_type .. ':' .. queue_id) ~= 1 then
3035
-- there are no more jobs in this queue. we can safely delete the interval.
31-
redis.call('HDEL', KEYS[1] .. ':interval', KEYS[2] .. ':' .. ARGV[1])
36+
redis.call('HDEL', prefix .. ':interval', queue_type .. ':' .. queue_id)
3237
end
3338

3439
-- delete the requeues_remaining entry for this job.
35-
redis.call('HDEL', KEYS[1] .. ':' .. KEYS[2] .. ':' .. ARGV[1] .. ':requeues_remaining', ARGV[2])
40+
redis.call('HDEL', prefix .. ':' .. queue_type .. ':' .. queue_id .. ':requeues_remaining', job_id)
3641

3742
return 1

‎sharq/scripts/lua/requeue.lua‎

Lines changed: 22 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -9,20 +9,24 @@
99
-- output:
1010
-- {} or job_discard_list
1111

12+
local prefix = KEYS[1]
13+
local queue_type = KEYS[2]
14+
local current_timestamp = ARGV[1]
15+
1216
-- check if any of the jobs need to be retried
13-
local requeue_job_list = redis.call('ZRANGEBYSCORE', KEYS[1] .. ':' .. KEYS[2] .. ':active', 0, ARGV[1])
17+
local requeue_job_list = redis.call('ZRANGEBYSCORE', prefix .. ':' .. queue_type .. ':active', 0, current_timestamp)
1418
local job_discard_list = {}
1519
-- iterate over each job and requeue it.
1620
for _, job in pairs(requeue_job_list) do
1721
local requeue = true
1822
local queue_id, job_id = job:match("([^,]+):([^,]+)")
1923
-- check if the job has any pending requeues.
20-
local requeues_remaining = redis.call('HGET', KEYS[1] .. ':' .. KEYS[2] .. ':' .. queue_id .. ':requeues_remaining', job_id)
21-
if tonumber(requeues_remaining) > -1 then
24+
local requeues_remaining = redis.call('HGET', prefix .. ':' .. queue_type .. ':' .. queue_id .. ':requeues_remaining', job_id)
25+
if requeues_remaining and tonumber(requeues_remaining) > -1 then
2226
-- finite requeues_remaining. decrement by one and check.
2327
requeues_remaining = requeues_remaining - 1
2428
-- update the new requeues_remaining value.
25-
redis.call('HSET', KEYS[1] .. ':' .. KEYS[2] .. ':' .. queue_id .. ':requeues_remaining', job_id, requeues_remaining)
29+
redis.call('HSET', prefix .. ':' .. queue_type .. ':' .. queue_id .. ':requeues_remaining', job_id, requeues_remaining)
2630
if requeues_remaining == -1 then
2731
-- discard this job
2832
table.insert(job_discard_list, job)
@@ -32,31 +36,31 @@ for _, job in pairs(requeue_job_list) do
3236
end
3337
if requeue == true then
3438
-- enqueue the job at the front of the job queue
35-
local job_queue_key = KEYS[1] .. ':' .. KEYS[2] .. ':' .. queue_id
39+
local job_queue_key = prefix .. ':' .. queue_type .. ':' .. queue_id
3640
redis.call('LPUSH', job_queue_key, job_id)
3741
-- check if this is the only job in the job queue
3842
if redis.call('LLEN', job_queue_key) == 1 then
43+
-- default when time keeper does not exist. next ready time is now.
44+
local next_ready_time = current_timestamp
3945
-- check if the time keeper exists
40-
local next_ready_time = 0
41-
if redis.call('EXISTS', KEYS[1] .. ':' .. KEYS[2] .. ':' .. queue_id .. ':time') == 1 then
42-
local last_dequeue_time = redis.call('GET', KEYS[1] .. ':' .. KEYS[2] .. ':' .. queue_id .. ':time')
43-
local interval = redis.call('HGET', KEYS[1] .. ':interval', KEYS[2] .. ':' .. queue_id)
46+
if redis.call('EXISTS', prefix .. ':' .. queue_type .. ':' .. queue_id .. ':time') == 1 then
47+
local last_dequeue_time = tonumber(redis.call('GET', prefix .. ':' .. queue_type .. ':' .. queue_id .. ':time'))
48+
local interval = tonumber(redis.call('HGET', prefix .. ':interval', queue_type .. ':' .. queue_id))
4449
-- compute next ready time
45-
next_ready_time = last_dequeue_time + interval
46-
else
47-
-- time keeper does not exist. next ready time is now.
48-
next_ready_time = ARGV[1]
50+
if last_dequeue_time and interval then
51+
next_ready_time = last_dequeue_time + interval
52+
end
4953
end
5054
-- insert this queue into the ready sorted set.
51-
redis.call('ZADD', KEYS[1] .. ':' .. KEYS[2], next_ready_time, queue_id)
52-
redis.call('SADD', KEYS[1] .. ':ready:queue_type', KEYS[2])
55+
redis.call('ZADD', prefix .. ':' .. queue_type, next_ready_time, queue_id)
56+
redis.call('SADD', prefix .. ':ready:queue_type', queue_type)
5357
end
5458
-- remove this queue_id & job_id from active sorted set.
55-
redis.call('ZREM', KEYS[1] .. ':' .. KEYS[2] .. ':active', queue_id .. ':' .. job_id)
59+
redis.call('ZREM', prefix .. ':' .. queue_type .. ':active', queue_id .. ':' .. job_id)
5660
-- check if the removed queue_id was the last item in this active set.
57-
if redis.call('EXISTS', KEYS[1] .. ':' .. KEYS[2] .. ':active') ~= 1 then
61+
if redis.call('EXISTS', prefix .. ':' .. queue_type .. ':active') ~= 1 then
5862
-- the active set does not exist. remove it from the metrics active queue type set.
59-
redis.call('SREM', KEYS[1] .. ':active:queue_type', KEYS[2])
63+
redis.call('SREM', prefix .. ':active:queue_type', queue_type)
6064
end
6165
end
6266
end

‎sharq/sharq.conf‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
[sharq]
2-
job_expire_interval = 1000
3-
job_requeue_interval = 1000
4-
default_job_requeue_limit = -1
2+
job_expire_interval : 120000 ; in milliseconds
3+
job_requeue_interval : 5000 ; in milliseconds
4+
default_job_requeue_limit : 0 ; value of -1 retries infinitely
55

66
[redis]
77
db = 0
@@ -10,4 +10,4 @@ conn_type = tcp_sock
1010
unix_socket_path = /tmp/redis.sock
1111
port = 6379
1212
host = 127.0.0.1
13-
clustered = false
13+
clustered = false

‎sharq/utils.py‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -65,6 +65,10 @@ def serialize_payload(payload):
6565
def deserialize_payload(payload):
6666
"""Tries to deserialize the payload using msgpack.
6767
"""
68+
# Handle older SharQ payloads as well (before py3 migration)
69+
if payload.startswith(b'"') and payload.endswith(b'"'):
70+
return msgpack.unpackb(payload[1:-1], raw=False)
71+
6872
return msgpack.unpackb(payload, raw=False)
6973

7074

@@ -83,4 +87,4 @@ def convert_to_str(queue_set):
8387
except Exception as e:
8488
queue_list.append(queue)
8589
pass
86-
return queue_list
90+
return queue_list

0 commit comments

Comments
 (0)