2
0
mirror of https://github.com/hibiken/asynq.git synced 2024-12-25 15:22:18 +08:00
asynq/recoverer_test.go
2022-04-11 16:55:43 -07:00

277 lines
7.5 KiB
Go

// Copyright 2020 Kentaro Hibino. All rights reserved.
// Use of this source code is governed by a MIT license
// that can be found in the LICENSE file.
package asynq
import (
"sync"
"testing"
"time"
"github.com/google/go-cmp/cmp"
"github.com/hibiken/asynq/internal/base"
"github.com/hibiken/asynq/internal/rdb"
h "github.com/hibiken/asynq/internal/testutil"
)
func TestRecoverer(t *testing.T) {
r := setup(t)
defer r.Close()
rdbClient := rdb.NewRDB(r)
t1 := h.NewTaskMessageWithQueue("task1", nil, "default")
t2 := h.NewTaskMessageWithQueue("task2", nil, "default")
t3 := h.NewTaskMessageWithQueue("task3", nil, "critical")
t4 := h.NewTaskMessageWithQueue("task4", nil, "default")
t4.Retried = t4.Retry // t4 has reached its max retry count
now := time.Now()
tests := []struct {
desc string
active map[string][]*base.TaskMessage
lease map[string][]base.Z
retry map[string][]base.Z
archived map[string][]base.Z
wantActive map[string][]*base.TaskMessage
wantLease map[string][]base.Z
wantRetry map[string][]*base.TaskMessage
wantArchived map[string][]*base.TaskMessage
}{
{
desc: "with one active task",
active: map[string][]*base.TaskMessage{
"default": {t1},
},
lease: map[string][]base.Z{
"default": {{Message: t1, Score: now.Add(-1 * time.Minute).Unix()}},
},
retry: map[string][]base.Z{
"default": {},
},
archived: map[string][]base.Z{
"default": {},
},
wantActive: map[string][]*base.TaskMessage{
"default": {},
},
wantLease: map[string][]base.Z{
"default": {},
},
wantRetry: map[string][]*base.TaskMessage{
"default": {t1},
},
wantArchived: map[string][]*base.TaskMessage{
"default": {},
},
},
{
desc: "with a task with max-retry reached",
active: map[string][]*base.TaskMessage{
"default": {t4},
"critical": {},
},
lease: map[string][]base.Z{
"default": {{Message: t4, Score: now.Add(-40 * time.Second).Unix()}},
"critical": {},
},
retry: map[string][]base.Z{
"default": {},
"critical": {},
},
archived: map[string][]base.Z{
"default": {},
"critical": {},
},
wantActive: map[string][]*base.TaskMessage{
"default": {},
"critical": {},
},
wantLease: map[string][]base.Z{
"default": {},
"critical": {},
},
wantRetry: map[string][]*base.TaskMessage{
"default": {},
"critical": {},
},
wantArchived: map[string][]*base.TaskMessage{
"default": {t4},
"critical": {},
},
},
{
desc: "with multiple active tasks, and one expired",
active: map[string][]*base.TaskMessage{
"default": {t1, t2},
"critical": {t3},
},
lease: map[string][]base.Z{
"default": {
{Message: t1, Score: now.Add(-2 * time.Minute).Unix()},
{Message: t2, Score: now.Add(20 * time.Second).Unix()},
},
"critical": {
{Message: t3, Score: now.Add(20 * time.Second).Unix()},
},
},
retry: map[string][]base.Z{
"default": {},
"critical": {},
},
archived: map[string][]base.Z{
"default": {},
"critical": {},
},
wantActive: map[string][]*base.TaskMessage{
"default": {t2},
"critical": {t3},
},
wantLease: map[string][]base.Z{
"default": {{Message: t2, Score: now.Add(20 * time.Second).Unix()}},
"critical": {{Message: t3, Score: now.Add(20 * time.Second).Unix()}},
},
wantRetry: map[string][]*base.TaskMessage{
"default": {t1},
"critical": {},
},
wantArchived: map[string][]*base.TaskMessage{
"default": {},
"critical": {},
},
},
{
desc: "with multiple expired active tasks",
active: map[string][]*base.TaskMessage{
"default": {t1, t2},
"critical": {t3},
},
lease: map[string][]base.Z{
"default": {
{Message: t1, Score: now.Add(-1 * time.Minute).Unix()},
{Message: t2, Score: now.Add(10 * time.Second).Unix()},
},
"critical": {
{Message: t3, Score: now.Add(-1 * time.Minute).Unix()},
},
},
retry: map[string][]base.Z{
"default": {},
"cricial": {},
},
archived: map[string][]base.Z{
"default": {},
"cricial": {},
},
wantActive: map[string][]*base.TaskMessage{
"default": {t2},
"critical": {},
},
wantLease: map[string][]base.Z{
"default": {{Message: t2, Score: now.Add(10 * time.Second).Unix()}},
},
wantRetry: map[string][]*base.TaskMessage{
"default": {t1},
"critical": {t3},
},
wantArchived: map[string][]*base.TaskMessage{
"default": {},
"critical": {},
},
},
{
desc: "with empty active queue",
active: map[string][]*base.TaskMessage{
"default": {},
"critical": {},
},
lease: map[string][]base.Z{
"default": {},
"critical": {},
},
retry: map[string][]base.Z{
"default": {},
"critical": {},
},
archived: map[string][]base.Z{
"default": {},
"critical": {},
},
wantActive: map[string][]*base.TaskMessage{
"default": {},
"critical": {},
},
wantLease: map[string][]base.Z{
"default": {},
"critical": {},
},
wantRetry: map[string][]*base.TaskMessage{
"default": {},
"critical": {},
},
wantArchived: map[string][]*base.TaskMessage{
"default": {},
"critical": {},
},
},
}
for _, tc := range tests {
h.FlushDB(t, r)
h.SeedAllActiveQueues(t, r, tc.active)
h.SeedAllLease(t, r, tc.lease)
h.SeedAllRetryQueues(t, r, tc.retry)
h.SeedAllArchivedQueues(t, r, tc.archived)
recoverer := newRecoverer(recovererParams{
logger: testLogger,
broker: rdbClient,
queues: []string{"default", "critical"},
interval: 1 * time.Second,
retryDelayFunc: func(n int, err error, task *Task) time.Duration { return 30 * time.Second },
isFailureFunc: defaultIsFailureFunc,
})
var wg sync.WaitGroup
recoverer.start(&wg)
runTime := time.Now() // time when recoverer is running
time.Sleep(2 * time.Second)
recoverer.shutdown()
for qname, want := range tc.wantActive {
gotActive := h.GetActiveMessages(t, r, qname)
if diff := cmp.Diff(want, gotActive, h.SortMsgOpt); diff != "" {
t.Errorf("%s; mismatch found in %q; (-want,+got)\n%s", tc.desc, base.ActiveKey(qname), diff)
}
}
for qname, want := range tc.wantLease {
gotLease := h.GetLeaseEntries(t, r, qname)
if diff := cmp.Diff(want, gotLease, h.SortZSetEntryOpt); diff != "" {
t.Errorf("%s; mismatch found in %q; (-want,+got)\n%s", tc.desc, base.LeaseKey(qname), diff)
}
}
cmpOpt := h.EquateInt64Approx(2) // allow up to two-second difference in `LastFailedAt`
for qname, msgs := range tc.wantRetry {
gotRetry := h.GetRetryMessages(t, r, qname)
var wantRetry []*base.TaskMessage // Note: construct message here since `LastFailedAt` is relative to each test run
for _, msg := range msgs {
wantRetry = append(wantRetry, h.TaskMessageAfterRetry(*msg, ErrLeaseExpired.Error(), runTime))
}
if diff := cmp.Diff(wantRetry, gotRetry, h.SortMsgOpt, cmpOpt); diff != "" {
t.Errorf("%s; mismatch found in %q: (-want, +got)\n%s", tc.desc, base.RetryKey(qname), diff)
}
}
for qname, msgs := range tc.wantArchived {
gotArchived := h.GetArchivedMessages(t, r, qname)
var wantArchived []*base.TaskMessage
for _, msg := range msgs {
wantArchived = append(wantArchived, h.TaskMessageWithError(*msg, ErrLeaseExpired.Error(), runTime))
}
if diff := cmp.Diff(wantArchived, gotArchived, h.SortMsgOpt, cmpOpt); diff != "" {
t.Errorf("%s; mismatch found in %q: (-want, +got)\n%s", tc.desc, base.ArchivedKey(qname), diff)
}
}
}
}