go-pkg/asynq
2
0
Fork 0
mirror of https://github.com/hibiken/asynq.git synced 2024-11-10 11:31:58 +08:00
Code Issues Projects Releases Wiki Activity

Update Retry method in RDB

Browse Source
This commit is contained in:
Ken Hibino 2020-08-08 12:17:33 -07:00
parent f63dcce0c0
commit 35b2b1782e
2 changed files with 100 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
internal/rdb/rdb.go
View File

@ -309,11 +309,11 @@ func (r *RDB) ScheduleUnique(msg *base.TaskMessage, processAt time.Time, ttl tim
return nil return nil
} }
// KEYS[1] -> asynq:in_progress // KEYS[1] -> asynq:{<qname>}:in_progress
// KEYS[2] -> asynq:deadlines // KEYS[2] -> asynq:{<qname>}:deadlines
// KEYS[3] -> asynq:retry // KEYS[3] -> asynq:{<qname>}:retry
// KEYS[4] -> asynq:processed:<yyyy-mm-dd> // KEYS[4] -> asynq:{<qname>}:processed:<yyyy-mm-dd>
// KEYS[5] -> asynq:failure:<yyyy-mm-dd> // KEYS[5] -> asynq:{<qname>}:failure:<yyyy-mm-dd>
// ARGV[1] -> base.TaskMessage value to remove from base.InProgressQueue queue // ARGV[1] -> base.TaskMessage value to remove from base.InProgressQueue queue
// ARGV[2] -> base.TaskMessage value to add to Retry queue // ARGV[2] -> base.TaskMessage value to add to Retry queue
// ARGV[3] -> retry_at UNIX timestamp // ARGV[3] -> retry_at UNIX timestamp
@ -351,11 +351,11 @@ func (r *RDB) Retry(msg *base.TaskMessage, processAt time.Time, errMsg string) e
return err return err
} }
now := time.Now() now := time.Now()
processedKey := base.ProcessedKey(now) processedKey := base.ProcessedKey(msg.Queue, now)
failureKey := base.FailureKey(now) failedKey := base.FailedKey(msg.Queue, now)
expireAt := now.Add(statsTTL) expireAt := now.Add(statsTTL)
return retryCmd.Run(r.client, return retryCmd.Run(r.client,
[]string{base.InProgressQueue, base.KeyDeadlines, base.RetryQueue, processedKey, failureKey}, []string{base.InProgressKey(msg.Queue), base.DeadlinesKey(msg.Queue), base.RetryKey(msg.Queue), processedKey, failedKey},
msgToRemove, msgToAdd, processAt.Unix(), expireAt.Unix()).Err() msgToRemove, msgToAdd, processAt.Unix(), expireAt.Unix()).Err()
} }

140
internal/rdb/rdb_test.go
View File

@ -779,61 +779,99 @@ func TestRetry(t *testing.T) {
Payload: map[string]interface{}{"subject": "Hola!"}, Payload: map[string]interface{}{"subject": "Hola!"},
Retried: 10, Retried: 10,
Timeout: 1800, Timeout: 1800,
Queue: "default",
} }
t1Deadline := now.Unix() + t1.Timeout
t2 := &base.TaskMessage{ t2 := &base.TaskMessage{
ID: uuid.New(), ID: uuid.New(),
Type: "gen_thumbnail", Type: "gen_thumbnail",
Payload: map[string]interface{}{"path": "some/path/to/image.jpg"}, Payload: map[string]interface{}{"path": "some/path/to/image.jpg"},
Timeout: 3000, Timeout: 3000,
Queue: "default",
} }
t2Deadline := now.Unix() + t2.Timeout
t3 := &base.TaskMessage{ t3 := &base.TaskMessage{
ID: uuid.New(), ID: uuid.New(),
Type: "reindex", Type: "reindex",
Payload: nil, Payload: nil,
Timeout: 60, Timeout: 60,
Queue: "default",
} }
t4 := &base.TaskMessage{
ID: uuid.New(),
Type: "send_notification",
Payload: nil,
Timeout: 1800,
Queue: "custom",
}
t1Deadline := now.Unix() + t1.Timeout
t2Deadline := now.Unix() + t2.Timeout
t4Deadline := now.Unix() + t4.Timeout
errMsg := "SMTP server is not responding" errMsg := "SMTP server is not responding"
tests := []struct { tests := []struct {
inProgress []*base.TaskMessage inProgress map[string][]*base.TaskMessage
deadlines []base.Z deadlines map[string][]base.Z
retry []base.Z retry map[string][]base.Z
msg *base.TaskMessage msg *base.TaskMessage
processAt time.Time processAt time.Time
errMsg string errMsg string
wantInProgress []*base.TaskMessage wantInProgress map[string][]*base.TaskMessage
wantDeadlines []base.Z wantDeadlines map[string][]base.Z
wantRetry []base.Z wantRetry map[string][]base.Z
}{ }{
{ {
inProgress: []*base.TaskMessage{t1, t2}, inProgress: map[string][]*base.TaskMessage{
deadlines: []base.Z{ "default": {t1, t2},
{Message: t1, Score: t1Deadline},
{Message: t2, Score: t2Deadline},
}, },
retry: []base.Z{ deadlines: map[string][]base.Z{
{ "default": {{Message: t1, Score: t1Deadline}, {Message: t2, Score: t2Deadline}},
Message: t3, },
Score: now.Add(time.Minute).Unix(), retry: map[string][]base.Z{
"default": {{Message: t3, Score: now.Add(time.Minute).Unix()}},
},
msg: t1,
processAt: now.Add(5 * time.Minute),
errMsg: errMsg,
wantInProgress: map[string][]*base.TaskMessage{
"default": {t2},
},
wantDeadlines: map[string][]base.Z{
"default": {{Message: t2, Score: t2Deadline}},
},
wantRetry: map[string][]base.Z{
"default": {
{Message: h.TaskMessageAfterRetry(*t1, errMsg), Score: now.Add(5 * time.Minute).Unix()},
{Message: t3, Score: now.Add(time.Minute).Unix()},
}, },
}, },
msg: t1, },
processAt: now.Add(5 * time.Minute), {
errMsg: errMsg, inProgress: map[string][]*base.TaskMessage{
wantInProgress: []*base.TaskMessage{t2}, "default": {t1, t2},
wantDeadlines: []base.Z{ "custom": {t4},
{Message: t2, Score: t2Deadline},
}, },
wantRetry: []base.Z{ deadlines: map[string][]base.Z{
{ "default": {{Message: t1, Score: t1Deadline}, {Message: t2, Score: t2Deadline}},
Message: h.TaskMessageAfterRetry(*t1, errMsg), "custom": {{Message: t4, Score: t4Deadline}},
Score: now.Add(5 * time.Minute).Unix(), },
}, retry: map[string][]base.Z{
{ "default": {},
Message: t3, "custom": {},
Score: now.Add(time.Minute).Unix(), },
msg: t4,
processAt: now.Add(5 * time.Minute),
errMsg: errMsg,
wantInProgress: map[string][]*base.TaskMessage{
"default": {t1, t2},
"custom": {},
},
wantDeadlines: map[string][]base.Z{
"default": {{Message: t1, Score: t1Deadline}, {Message: t2, Score: t2Deadline}},
"custom": {},
},
wantRetry: map[string][]base.Z{
"default": {},
"custom": {
{Message: h.TaskMessageAfterRetry(*t4, errMsg), Score: now.Add(5 * time.Minute).Unix()},
}, },
}, },
}, },
@ -841,9 +879,9 @@ func TestRetry(t *testing.T) {
for _, tc := range tests { for _, tc := range tests {
h.FlushDB(t, r.client) h.FlushDB(t, r.client)
h.SeedInProgressQueue(t, r.client, tc.inProgress) h.SeedAllInProgressQueues(t, r.client, tc.inProgress)
h.SeedDeadlines(t, r.client, tc.deadlines) h.SeedAllDeadlines(t, r.client, tc.deadlines)
h.SeedRetryQueue(t, r.client, tc.retry) h.SeedAllRetryQueues(t, r.client, tc.retry)
err := r.Retry(tc.msg, tc.processAt, tc.errMsg) err := r.Retry(tc.msg, tc.processAt, tc.errMsg)
if err != nil { if err != nil {
@ -851,20 +889,26 @@ func TestRetry(t *testing.T) {
continue continue
} }
gotInProgress := h.GetInProgressMessages(t, r.client) for queue, want := range tc.wantInProgress {
if diff := cmp.Diff(tc.wantInProgress, gotInProgress, h.SortMsgOpt); diff != "" { gotInProgress := h.GetInProgressMessages(t, r.client, queue)
t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.InProgressQueue, diff) if diff := cmp.Diff(want, gotInProgress, h.SortMsgOpt); diff != "" {
t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.InProgressKey(queue), diff)
}
} }
gotDeadlines := h.GetDeadlinesEntries(t, r.client) for queue, want := range tc.wantDeadlines {
if diff := cmp.Diff(tc.wantDeadlines, gotDeadlines, h.SortZSetEntryOpt); diff != "" { gotDeadlines := h.GetDeadlinesEntries(t, r.client, queue)
t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.KeyDeadlines, diff) if diff := cmp.Diff(want, gotDeadlines, h.SortZSetEntryOpt); diff != "" {
t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.DeadlinesKey(queue), diff)
}
} }
gotRetry := h.GetRetryEntries(t, r.client) for queue, want := range tc.wantRetry {
if diff := cmp.Diff(tc.wantRetry, gotRetry, h.SortZSetEntryOpt); diff != "" { gotRetry := h.GetRetryEntries(t, r.client, queue)
t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.RetryQueue, diff) if diff := cmp.Diff(want, gotRetry, h.SortZSetEntryOpt); diff != "" {
t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.RetryKey(queue), diff)
}
} }
processedKey := base.ProcessedKey(time.Now()) processedKey := base.ProcessedKey(tc.msg.Queue, time.Now())
gotProcessed := r.client.Get(processedKey).Val() gotProcessed := r.client.Get(processedKey).Val()
if gotProcessed != "1" { if gotProcessed != "1" {
t.Errorf("GET %q = %q, want 1", processedKey, gotProcessed) t.Errorf("GET %q = %q, want 1", processedKey, gotProcessed)
@ -874,14 +918,14 @@ func TestRetry(t *testing.T) {
t.Errorf("TTL %q = %v, want less than or equal to %v", processedKey, gotTTL, statsTTL) t.Errorf("TTL %q = %v, want less than or equal to %v", processedKey, gotTTL, statsTTL)
} }
failureKey := base.FailureKey(time.Now()) failedKey := base.FailedKey(tc.msg.Queue, time.Now())
gotFailure := r.client.Get(failureKey).Val() gotFailure := r.client.Get(failedKey).Val()
if gotFailure != "1" { if gotFailure != "1" {
t.Errorf("GET %q = %q, want 1", failureKey, gotFailure) t.Errorf("GET %q = %q, want 1", failedKey, gotFailure)
} }
gotTTL = r.client.TTL(processedKey).Val() gotTTL = r.client.TTL(failedKey).Val()
if gotTTL > statsTTL { if gotTTL > statsTTL {
t.Errorf("TTL %q = %v, want less than or equal to %v", failureKey, gotTTL, statsTTL) t.Errorf("TTL %q = %v, want less than or equal to %v", failedKey, gotTTL, statsTTL)
} }
} }
} }