mirror of
https://github.com/hibiken/asynq.git
synced 2024-12-25 23:32:17 +08:00
Update RDB.Retry to remove message from deadlines set
This commit is contained in:
parent
7b9119c703
commit
9cd9f3d6b4
@ -315,9 +315,10 @@ func (r *RDB) ScheduleUnique(msg *base.TaskMessage, processAt time.Time, ttl tim
|
||||
}
|
||||
|
||||
// KEYS[1] -> asynq:in_progress
|
||||
// KEYS[2] -> asynq:retry
|
||||
// KEYS[3] -> asynq:processed:<yyyy-mm-dd>
|
||||
// KEYS[4] -> asynq:failure:<yyyy-mm-dd>
|
||||
// KEYS[2] -> asynq:deadlines
|
||||
// KEYS[3] -> asynq:retry
|
||||
// KEYS[4] -> asynq:processed:<yyyy-mm-dd>
|
||||
// KEYS[5] -> asynq:failure:<yyyy-mm-dd>
|
||||
// ARGV[1] -> base.TaskMessage value to remove from base.InProgressQueue queue
|
||||
// ARGV[2] -> base.TaskMessage value to add to Retry queue
|
||||
// ARGV[3] -> retry_at UNIX timestamp
|
||||
@ -327,15 +328,19 @@ local x = redis.call("LREM", KEYS[1], 0, ARGV[1])
|
||||
if x == 0 then
|
||||
return redis.error_reply("NOT FOUND")
|
||||
end
|
||||
redis.call("ZADD", KEYS[2], ARGV[3], ARGV[2])
|
||||
local n = redis.call("INCR", KEYS[3])
|
||||
if tonumber(n) == 1 then
|
||||
redis.call("EXPIREAT", KEYS[3], ARGV[4])
|
||||
x = redis.call("ZREM", KEYS[2], ARGV[1])
|
||||
if x == 0 then
|
||||
return redis.error_reply("NOT FOUND")
|
||||
end
|
||||
local m = redis.call("INCR", KEYS[4])
|
||||
if tonumber(m) == 1 then
|
||||
redis.call("ZADD", KEYS[3], ARGV[3], ARGV[2])
|
||||
local n = redis.call("INCR", KEYS[4])
|
||||
if tonumber(n) == 1 then
|
||||
redis.call("EXPIREAT", KEYS[4], ARGV[4])
|
||||
end
|
||||
local m = redis.call("INCR", KEYS[5])
|
||||
if tonumber(m) == 1 then
|
||||
redis.call("EXPIREAT", KEYS[5], ARGV[4])
|
||||
end
|
||||
return redis.status_reply("OK")`)
|
||||
|
||||
// Retry moves the task from in-progress to retry queue, incrementing retry count
|
||||
@ -357,7 +362,7 @@ func (r *RDB) Retry(msg *base.TaskMessage, processAt time.Time, errMsg string) e
|
||||
failureKey := base.FailureKey(now)
|
||||
expireAt := now.Add(statsTTL)
|
||||
return retryCmd.Run(r.client,
|
||||
[]string{base.InProgressQueue, base.RetryQueue, processedKey, failureKey},
|
||||
[]string{base.InProgressQueue, base.KeyDeadlines, base.RetryQueue, processedKey, failureKey},
|
||||
msgToRemove, msgToAdd, processAt.Unix(), expireAt.Unix()).Err()
|
||||
}
|
||||
|
||||
|
@ -694,9 +694,27 @@ func TestScheduleUnique(t *testing.T) {
|
||||
|
||||
func TestRetry(t *testing.T) {
|
||||
r := setup(t)
|
||||
t1 := h.NewTaskMessage("send_email", map[string]interface{}{"subject": "Hola!"})
|
||||
t2 := h.NewTaskMessage("gen_thumbnail", map[string]interface{}{"path": "some/path/to/image.jpg"})
|
||||
t3 := h.NewTaskMessage("reindex", nil)
|
||||
now := time.Now()
|
||||
t1 := &base.TaskMessage{
|
||||
ID: xid.New(),
|
||||
Type: "send_email",
|
||||
Payload: map[string]interface{}{"subject": "Hola!"},
|
||||
Timeout: 1800,
|
||||
}
|
||||
t1Deadline := int(now.Unix()) + t1.Timeout
|
||||
t2 := &base.TaskMessage{
|
||||
ID: xid.New(),
|
||||
Type: "gen_thumbnail",
|
||||
Payload: map[string]interface{}{"path": "some/path/to/image.jpg"},
|
||||
Timeout: 3000,
|
||||
}
|
||||
t2Deadline := int(now.Unix()) + t2.Timeout
|
||||
t3 := &base.TaskMessage{
|
||||
ID: xid.New(),
|
||||
Type: "reindex",
|
||||
Payload: nil,
|
||||
Timeout: 60,
|
||||
}
|
||||
t1.Retried = 10
|
||||
errMsg := "SMTP server is not responding"
|
||||
t1AfterRetry := &base.TaskMessage{
|
||||
@ -706,21 +724,27 @@ func TestRetry(t *testing.T) {
|
||||
Queue: t1.Queue,
|
||||
Retry: t1.Retry,
|
||||
Retried: t1.Retried + 1,
|
||||
Timeout: t1.Timeout,
|
||||
ErrorMsg: errMsg,
|
||||
}
|
||||
now := time.Now()
|
||||
|
||||
tests := []struct {
|
||||
inProgress []*base.TaskMessage
|
||||
deadlines []h.ZSetEntry
|
||||
retry []h.ZSetEntry
|
||||
msg *base.TaskMessage
|
||||
processAt time.Time
|
||||
errMsg string
|
||||
wantInProgress []*base.TaskMessage
|
||||
wantDeadlines []h.ZSetEntry
|
||||
wantRetry []h.ZSetEntry
|
||||
}{
|
||||
{
|
||||
inProgress: []*base.TaskMessage{t1, t2},
|
||||
deadlines: []h.ZSetEntry{
|
||||
{Msg: t1, Score: float64(t1Deadline)},
|
||||
{Msg: t2, Score: float64(t2Deadline)},
|
||||
},
|
||||
retry: []h.ZSetEntry{
|
||||
{
|
||||
Msg: t3,
|
||||
@ -731,6 +755,9 @@ func TestRetry(t *testing.T) {
|
||||
processAt: now.Add(5 * time.Minute),
|
||||
errMsg: errMsg,
|
||||
wantInProgress: []*base.TaskMessage{t2},
|
||||
wantDeadlines: []h.ZSetEntry{
|
||||
{Msg: t2, Score: float64(t2Deadline)},
|
||||
},
|
||||
wantRetry: []h.ZSetEntry{
|
||||
{
|
||||
Msg: t1AfterRetry,
|
||||
@ -747,6 +774,7 @@ func TestRetry(t *testing.T) {
|
||||
for _, tc := range tests {
|
||||
h.FlushDB(t, r.client)
|
||||
h.SeedInProgressQueue(t, r.client, tc.inProgress)
|
||||
h.SeedDeadlines(t, r.client, tc.deadlines)
|
||||
h.SeedRetryQueue(t, r.client, tc.retry)
|
||||
|
||||
err := r.Retry(tc.msg, tc.processAt, tc.errMsg)
|
||||
@ -759,7 +787,10 @@ func TestRetry(t *testing.T) {
|
||||
if diff := cmp.Diff(tc.wantInProgress, gotInProgress, h.SortMsgOpt); diff != "" {
|
||||
t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.InProgressQueue, diff)
|
||||
}
|
||||
|
||||
gotDeadlines := h.GetDeadlinesEntries(t, r.client)
|
||||
if diff := cmp.Diff(tc.wantDeadlines, gotDeadlines, h.SortZSetEntryOpt); diff != "" {
|
||||
t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.KeyDeadlines, diff)
|
||||
}
|
||||
gotRetry := h.GetRetryEntries(t, r.client)
|
||||
if diff := cmp.Diff(tc.wantRetry, gotRetry, h.SortZSetEntryOpt); diff != "" {
|
||||
t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.RetryQueue, diff)
|
||||
|
Loading…
Reference in New Issue
Block a user