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 Kill method in RDB

Browse Source
This commit is contained in:
Ken Hibino 2020-08-08 12:44:08 -07:00
parent 35b2b1782e
commit 74f08795f8
2 changed files with 130 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
internal/rdb/rdb.go
View File

@ -364,11 +364,11 @@ const (
deadExpirationInDays = 90 deadExpirationInDays = 90
) )
// KEYS[1] -> asynq:in_progress // KEYS[1] -> asynq:{<qname>}:in_progress
// KEYS[2] -> asynq:deadlines // KEYS[2] -> asynq:{<qname>}:deadlines
// KEYS[3] -> asynq:dead // KEYS[3] -> asynq:{<qname>}:dead
// 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 Dead queue // ARGV[2] -> base.TaskMessage value to add to Dead queue
// ARGV[3] -> died_at UNIX timestamp // ARGV[3] -> died_at UNIX timestamp
@ -411,11 +411,11 @@ func (r *RDB) Kill(msg *base.TaskMessage, errMsg string) error {
} }
now := time.Now() now := time.Now()
limit := now.AddDate(0, 0, -deadExpirationInDays).Unix() // 90 days ago limit := now.AddDate(0, 0, -deadExpirationInDays).Unix() // 90 days ago
processedKey := base.ProcessedKey(now) processedKey := base.ProcessedKey(msg.Queue, now)
failureKey := base.FailureKey(now) failureKey := base.FailureKey(msg.Queue, now)
expireAt := now.Add(statsTTL) expireAt := now.Add(statsTTL)
return killCmd.Run(r.client, return killCmd.Run(r.client,
[]string{base.InProgressQueue, base.KeyDeadlines, base.DeadQueue, processedKey, failureKey}, []string{base.InProgressKey(msg.Queue), base.DeadlinesKey(msg.Queue), base.DeadKey(msg.Queue), processedKey, failureKey},
msgToRemove, msgToAdd, now.Unix(), limit, maxDeadTasks, expireAt.Unix()).Err() msgToRemove, msgToAdd, now.Unix(), limit, maxDeadTasks, expireAt.Unix()).Err()
} }

185
internal/rdb/rdb_test.go
View File

@ -919,9 +919,9 @@ func TestRetry(t *testing.T) {
} }
failedKey := base.FailedKey(tc.msg.Queue, time.Now()) failedKey := base.FailedKey(tc.msg.Queue, time.Now())
gotFailure := r.client.Get(failedKey).Val() gotFailed := r.client.Get(failedKey).Val()
if gotFailure != "1" { if gotFailed != "1" {
t.Errorf("GET %q = %q, want 1", failedKey, gotFailure) t.Errorf("GET %q = %q, want 1", failedKey, gotFailed)
} }
gotTTL = r.client.TTL(failedKey).Val() gotTTL = r.client.TTL(failedKey).Val()
if gotTTL > statsTTL { if gotTTL > statsTTL {
@ -939,7 +939,7 @@ func TestKill(t *testing.T) {
Payload: nil, Payload: nil,
Queue: "default", Queue: "default",
Retry: 25, Retry: 25,
Retried: 0, Retried: 25,
Timeout: 1800, Timeout: 1800,
} }
t1Deadline := now.Unix() + t1.Timeout t1Deadline := now.Unix() + t1.Timeout
@ -963,64 +963,117 @@ func TestKill(t *testing.T) {
Timeout: 60, Timeout: 60,
} }
t3Deadline := now.Unix() + t3.Timeout t3Deadline := now.Unix() + t3.Timeout
t4 := &base.TaskMessage{
ID: uuid.New(),
Type: "send_email",
Payload: nil,
Queue: "custom",
Retry: 25,
Retried: 25,
Timeout: 1800,
}
t4Deadline := now.Unix() + t4.Timeout
errMsg := "SMTP server not responding" errMsg := "SMTP server not responding"
// TODO(hibiken): add test cases for trimming // TODO(hibiken): add test cases for trimming
tests := []struct { tests := []struct {
inProgress []*base.TaskMessage inProgress map[string][]*base.TaskMessage
deadlines []base.Z deadlines map[string][]base.Z
dead []base.Z dead map[string][]base.Z
target *base.TaskMessage // task to kill target *base.TaskMessage // task to kill
wantInProgress []*base.TaskMessage wantInProgress map[string][]*base.TaskMessage
wantDeadlines []base.Z wantDeadlines map[string][]base.Z
wantDead []base.Z wantDead 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},
}, },
dead: []base.Z{ deadlines: map[string][]base.Z{
{ "default": {
Message: t3, {Message: t1, Score: t1Deadline},
Score: now.Add(-time.Hour).Unix(), {Message: t2, Score: t2Deadline},
}, },
}, },
target: t1, dead: map[string][]base.Z{
wantInProgress: []*base.TaskMessage{t2}, "default": {
wantDeadlines: []base.Z{ {Message: t3, Score: now.Add(-time.Hour).Unix()},
{Message: t2, Score: t2Deadline},
},
wantDead: []base.Z{
{
Message: h.TaskMessageWithError(*t1, errMsg),
Score: now.Unix(),
}, },
{ },
Message: t3, target: t1,
Score: now.Add(-time.Hour).Unix(), wantInProgress: map[string][]*base.TaskMessage{
"default": {t2},
},
wantDeadlines: map[string]base.Z{
"default": {{Message: t2, Score: t2Deadline}},
},
wantDead: map[string][]base.Z{
"default": {
{Message: h.TaskMessageWithError(*t1, errMsg), Score: now.Unix()},
{Message: t3, Score: now.Add(-time.Hour).Unix()},
}, },
}, },
}, },
{ {
inProgress: []*base.TaskMessage{t1, t2, t3}, inProgress: map[string][]*base.TaskMessage{
deadlines: []base.Z{ "default": {t1, t2, t3},
{Message: t1, Score: t1Deadline},
{Message: t2, Score: t2Deadline},
{Message: t3, Score: t3Deadline},
}, },
dead: []base.Z{}, deadlines: map[string][]base.Z{
target: t1, "default": {
wantInProgress: []*base.TaskMessage{t2, t3}, {Message: t1, Score: t1Deadline},
wantDeadlines: []base.Z{ {Message: t2, Score: t2Deadline},
{Message: t2, Score: t2Deadline}, {Message: t3, Score: t3Deadline},
{Message: t3, Score: t3Deadline}, },
}, },
wantDead: []base.Z{ dead: map[string][]base.Z{
{ "default": {},
Message: h.TaskMessageWithError(*t1, errMsg), },
Score: now.Unix(), target: t1,
wantInProgress: map[string][]*base.TaskMessage{
"default": {t2, t3},
},
wantDeadlines: map[string][]base.Z{
"default": {
{Message: t2, Score: t2Deadline},
{Message: t3, Score: t3Deadline},
},
},
wantDead: map[string][]base.Z{
"default": {
{Message: h.TaskMessageWithError(*t1, errMsg), Score: now.Unix()},
},
},
},
{
inProgress: map[string][]*base.TaskMessage{
"default": {t1},
"custom": {t4},
},
deadlines: map[string][]base.Z{
"default": {
{Message: t1, Score: t1Deadline},
},
"custom": {
{Message: t4, Score: t4Deadline},
},
},
dead: map[string][]base.Z{
"default": {},
"custom": {},
},
target: t4,
wantInProgress: map[string][]*base.TaskMessage{
"default": {t1},
"custom": {},
},
wantDeadlines: map[string]base.Z{
"default": {{Message: t1, Score: t1Deadline}},
"custom": {},
},
wantDead: map[string][]base.Z{
"default": {},
"custom": {
{Message: h.TaskMessageWithError(*t4, errMsg), Score: now.Unix()},
}, },
}, },
}, },
@ -1028,9 +1081,9 @@ func TestKill(t *testing.T) {
for _, tc := range tests { for _, tc := range tests {
h.FlushDB(t, r.client) // clean up db before each test case h.FlushDB(t, r.client) // clean up db before each test case
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.SeedDeadQueue(t, r.client, tc.dead) h.SeedAllDeadQueues(t, r.client, tc.dead)
err := r.Kill(tc.target, errMsg) err := r.Kill(tc.target, errMsg)
if err != nil { if err != nil {
@ -1038,20 +1091,26 @@ func TestKill(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 after calling (*RDB).Kill: (-want, +got):\n%s", base.KeyDeadlines, diff) if diff := cmp.Diff(want, gotDeadlines, h.SortZSetEntryOpt); diff != "" {
t.Errorf("mismatch found in %q after calling (*RDB).Kill: (-want, +got):\n%s", base.DeadlinesKey(queue), diff)
}
} }
gotDead := h.GetDeadEntries(t, r.client) for queue, want := range tc.wantDead {
if diff := cmp.Diff(tc.wantDead, gotDead, h.SortZSetEntryOpt); diff != "" { gotDead := h.GetDeadEntries(t, r.client, queue)
t.Errorf("mismatch found in %q after calling (*RDB).Kill: (-want, +got):\n%s", base.DeadQueue, diff) if diff := cmp.Diff(want, gotDead, h.SortZSetEntryOpt); diff != "" {
t.Errorf("mismatch found in %q after calling (*RDB).Kill: (-want, +got):\n%s", base.DeadKey(queue), diff)
}
} }
processedKey := base.ProcessedKey(time.Now()) processedKey := base.ProcessedKey(tc.target.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)
@ -1061,14 +1120,14 @@ func TestKill(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.target.Queue, time.Now())
gotFailure := r.client.Get(failureKey).Val() gotFailed := r.client.Get(failedKey).Val()
if gotFailure != "1" { if gotFailed != "1" {
t.Errorf("GET %q = %q, want 1", failureKey, gotFailure) t.Errorf("GET %q = %q, want 1", failedKey, gotFailed)
} }
gotTTL = r.client.TTL(processedKey).Val() gotTTL = r.client.TTL(processedKey).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)
} }
} }
} }