diff --git a/internal/rdb/inspect.go b/internal/rdb/inspect.go index 2a2bd67..3f6e856 100644 --- a/internal/rdb/inspect.go +++ b/internal/rdb/inspect.go @@ -743,77 +743,79 @@ func (r *RDB) archiveAll(src, dst, qname string) (int64, error) { return n, nil } -// DeleteArchivedTask deletes an archived task that matches the given id and score from the given queue. -// If a task that matches the id and score does not exist, it returns ErrTaskNotFound. -func (r *RDB) DeleteArchivedTask(qname string, id uuid.UUID) error { - return r.deleteTask(base.ArchivedKey(qname), qname, id.String()) -} - -// DeleteRetryTask deletes a retry task that matches the given id and score from the given queue. -// If a task that matches the id and score does not exist, it returns ErrTaskNotFound. -func (r *RDB) DeleteRetryTask(qname string, id uuid.UUID) error { - return r.deleteTask(base.RetryKey(qname), qname, id.String()) -} - -// DeleteScheduledTask deletes a scheduled task that matches the given id and score from the given queue. -// If a task that matches the id and score does not exist, it returns ErrTaskNotFound. -func (r *RDB) DeleteScheduledTask(qname string, id uuid.UUID) error { - return r.deleteTask(base.ScheduledKey(qname), qname, id.String()) -} - -// KEYS[1] -> asynq:{}:pending -// KEYS[2] -> asynq:{}:t: -// ARGV[1] -> task ID -var deletePendingTaskCmd = redis.NewScript(` -if redis.call("LREM", KEYS[1], 0, ARGV[1]) == 0 then - return 0 -end -return redis.call("DEL", KEYS[2]) -`) - -// DeletePendingTask deletes a pending tasks that matches the given id from the given queue. -// If there's no match, it returns ErrTaskNotFound. -func (r *RDB) DeletePendingTask(qname string, id uuid.UUID) error { - keys := []string{base.PendingKey(qname), base.TaskKey(qname, id.String())} - res, err := deletePendingTaskCmd.Run(r.client, keys, id.String()).Result() - if err != nil { - return err - } - n, ok := res.(int64) - if !ok { - return fmt.Errorf("command error: unexpected return value %v", res) - } - if n == 0 { - return ErrTaskNotFound - } - return nil -} - -// KEYS[1] -> ZSET key to remove the task from (e.g. asynq:{}:retry) -// KEYS[2] -> asynq:{}:t: +// Input: +// KEYS[1] -> asynq:{}:t: +// KEYS[2] -> all queues key // ARGV[1] -> task ID +// ARGV[2] -> queue key prefix +// ARGV[3] -> queue name +// +// Output: +// Numeric code indicating the status: +// Returns 1 if task is successfully deleted. +// Returns 0 if task is not found. +// Returns -1 if queue doesn't exist. +// Returns -2 if task is in active state. var deleteTaskCmd = redis.NewScript(` -if redis.call("ZREM", KEYS[1], ARGV[1]) == 0 then +if redis.call("SISMEMBER", KEYS[2], ARGV[3]) == 0 then + return -1 +end +if redis.call("EXISTS", KEYS[1]) == 0 then return 0 end -return redis.call("DEL", KEYS[2]) +local state = redis.call("HGET", KEYS[1], "state") +if state == "active" then + return -2 +end +if state == "pending" then + if redis.call("LREM", ARGV[2] .. state, 0, ARGV[1]) == 0 then + return redis.error_reply("task is not found in list: " .. tostring(state)) + end +else + if redis.call("ZREM", ARGV[2] .. state, ARGV[1]) == 0 then + return redis.error_reply("task is not found in zset: " .. tostring(state)) + end +end +return redis.call("DEL", KEYS[1]) `) -func (r *RDB) deleteTask(key, qname, id string) error { - keys := []string{key, base.TaskKey(qname, id)} - argv := []interface{}{id} +// DeleteTask finds a task that matches the id from the given queue and deletes it. +// It returns nil if it successfully archived the task. +// +// If a queue with the given name doesn't exist, it returns QueueNotFoundError. +// If a task with the given id doesn't exist in the queue, it returns TaskNotFoundError +// If a task is in active state it returns non-nil error with Code FailedPrecondition. +func (r *RDB) DeleteTask(qname string, id uuid.UUID) error { + var op errors.Op = "rdb.DeleteTask" + keys := []string{ + base.TaskKey(qname, id.String()), + base.AllQueues, + } + argv := []interface{}{ + id.String(), + base.QueueKeyPrefix(qname), + qname, + } res, err := deleteTaskCmd.Run(r.client, keys, argv...).Result() if err != nil { - return err + return errors.E(op, errors.Unknown, err) } n, ok := res.(int64) if !ok { - return fmt.Errorf("command error: unexpected return value %v", res) + return errors.E(op, errors.Internal, fmt.Sprintf("cast error: deleteTaskCmd script returned unexported value %v", res)) } - if n == 0 { - return ErrTaskNotFound + switch n { + case 1: + return nil + case 0: + return errors.E(op, errors.NotFound, &errors.TaskNotFoundError{Queue: qname, ID: id.String()}) + case -1: + return errors.E(op, errors.NotFound, &errors.QueueNotFoundError{Queue: qname}) + case -2: + return errors.E(op, errors.FailedPrecondition, "cannot delete task in active state. use CancelTask instead.") + default: + return errors.E(op, errors.Internal, fmt.Sprintf("unexpected return value from deleteTaskCmd script: %d", n)) } - return nil } // KEYS[1] -> queue to delete diff --git a/internal/rdb/inspect_test.go b/internal/rdb/inspect_test.go index 476c88d..0855d8c 100644 --- a/internal/rdb/inspect_test.go +++ b/internal/rdb/inspect_test.go @@ -2445,7 +2445,6 @@ func TestDeleteArchivedTask(t *testing.T) { archived map[string][]base.Z qname string id uuid.UUID - want error wantArchived map[string][]*base.TaskMessage }{ { @@ -2457,7 +2456,6 @@ func TestDeleteArchivedTask(t *testing.T) { }, qname: "default", id: m1.ID, - want: nil, wantArchived: map[string][]*base.TaskMessage{ "default": {m2}, }, @@ -2474,46 +2472,19 @@ func TestDeleteArchivedTask(t *testing.T) { }, qname: "custom", id: m3.ID, - want: nil, wantArchived: map[string][]*base.TaskMessage{ "default": {m1, m2}, "custom": {}, }, }, - { - archived: map[string][]base.Z{ - "default": { - {Message: m1, Score: t1.Unix()}, - {Message: m2, Score: t2.Unix()}, - }, - }, - qname: "default", - id: uuid.New(), - want: ErrTaskNotFound, - wantArchived: map[string][]*base.TaskMessage{ - "default": {m1, m2}, - }, - }, - { - archived: map[string][]base.Z{ - "default": {}, - }, - qname: "default", - id: m1.ID, - want: ErrTaskNotFound, - wantArchived: map[string][]*base.TaskMessage{ - "default": {}, - }, - }, } for _, tc := range tests { h.FlushDB(t, r.client) // clean up db before each test case h.SeedAllArchivedQueues(t, r.client, tc.archived) - got := r.DeleteArchivedTask(tc.qname, tc.id) - if got != tc.want { - t.Errorf("r.DeleteArchivedTask(%q, %v) = %v, want %v", tc.qname, tc.id, got, tc.want) + if got := r.DeleteTask(tc.qname, tc.id); got != nil { + t.Errorf("r.DeleteTask(%q, %v) returned error: %v", tc.qname, tc.id, got) continue } @@ -2540,7 +2511,6 @@ func TestDeleteRetryTask(t *testing.T) { retry map[string][]base.Z qname string id uuid.UUID - want error wantRetry map[string][]*base.TaskMessage }{ { @@ -2552,7 +2522,6 @@ func TestDeleteRetryTask(t *testing.T) { }, qname: "default", id: m1.ID, - want: nil, wantRetry: map[string][]*base.TaskMessage{ "default": {m2}, }, @@ -2569,32 +2538,19 @@ func TestDeleteRetryTask(t *testing.T) { }, qname: "custom", id: m3.ID, - want: nil, wantRetry: map[string][]*base.TaskMessage{ "default": {m1, m2}, "custom": {}, }, }, - { - retry: map[string][]base.Z{ - "default": {{Message: m1, Score: t1.Unix()}}, - }, - qname: "default", - id: uuid.New(), - want: ErrTaskNotFound, - wantRetry: map[string][]*base.TaskMessage{ - "default": {m1}, - }, - }, } for _, tc := range tests { h.FlushDB(t, r.client) // clean up db before each test case h.SeedAllRetryQueues(t, r.client, tc.retry) - got := r.DeleteRetryTask(tc.qname, tc.id) - if got != tc.want { - t.Errorf("r.DeleteRetryTask(%q, %v) = %v, want %v", tc.qname, tc.id, got, tc.want) + if got := r.DeleteTask(tc.qname, tc.id); got != nil { + t.Errorf("r.DeleteTask(%q, %v) returned error: %v", tc.qname, tc.id, got) continue } @@ -2621,7 +2577,6 @@ func TestDeleteScheduledTask(t *testing.T) { scheduled map[string][]base.Z qname string id uuid.UUID - want error wantScheduled map[string][]*base.TaskMessage }{ { @@ -2633,7 +2588,6 @@ func TestDeleteScheduledTask(t *testing.T) { }, qname: "default", id: m1.ID, - want: nil, wantScheduled: map[string][]*base.TaskMessage{ "default": {m2}, }, @@ -2650,32 +2604,19 @@ func TestDeleteScheduledTask(t *testing.T) { }, qname: "custom", id: m3.ID, - want: nil, wantScheduled: map[string][]*base.TaskMessage{ "default": {m1, m2}, "custom": {}, }, }, - { - scheduled: map[string][]base.Z{ - "default": {{Message: m1, Score: t1.Unix()}}, - }, - qname: "default", - id: uuid.New(), - want: ErrTaskNotFound, - wantScheduled: map[string][]*base.TaskMessage{ - "default": {m1}, - }, - }, } for _, tc := range tests { h.FlushDB(t, r.client) // clean up db before each test case h.SeedAllScheduledQueues(t, r.client, tc.scheduled) - got := r.DeleteScheduledTask(tc.qname, tc.id) - if got != tc.want { - t.Errorf("r.DeleteScheduledTask(%q, %v) = %v, want %v", tc.qname, tc.id, got, tc.want) + if got := r.DeleteTask(tc.qname, tc.id); got != nil { + t.Errorf("r.DeleteTask(%q, %v) returned error: %v", tc.qname, tc.id, got) continue } @@ -2699,7 +2640,6 @@ func TestDeletePendingTask(t *testing.T) { pending map[string][]*base.TaskMessage qname string id uuid.UUID - want error wantPending map[string][]*base.TaskMessage }{ { @@ -2708,7 +2648,6 @@ func TestDeletePendingTask(t *testing.T) { }, qname: "default", id: m1.ID, - want: nil, wantPending: map[string][]*base.TaskMessage{ "default": {m2}, }, @@ -2720,32 +2659,19 @@ func TestDeletePendingTask(t *testing.T) { }, qname: "custom", id: m3.ID, - want: nil, wantPending: map[string][]*base.TaskMessage{ "default": {m1, m2}, "custom": {}, }, }, - { - pending: map[string][]*base.TaskMessage{ - "default": {m1, m2}, - }, - qname: "default", - id: uuid.New(), - want: ErrTaskNotFound, - wantPending: map[string][]*base.TaskMessage{ - "default": {m1, m2}, - }, - }, } for _, tc := range tests { h.FlushDB(t, r.client) h.SeedAllPendingQueues(t, r.client, tc.pending) - got := r.DeletePendingTask(tc.qname, tc.id) - if got != tc.want { - t.Errorf("r.DeletePendingTask(%q, %v) = %v, want %v", tc.qname, tc.id, got, tc.want) + if got := r.DeleteTask(tc.qname, tc.id); got != nil { + t.Errorf("r.DeleteTask(%q, %v) returned error: %v", tc.qname, tc.id, got) continue } @@ -2758,6 +2684,65 @@ func TestDeletePendingTask(t *testing.T) { } } +func TestDeleteTaskError(t *testing.T) { + r := setup(t) + defer r.Close() + m1 := h.NewTaskMessage("task1", nil) + t1 := time.Now().Add(5 * time.Minute) + + tests := []struct { + desc string + scheduled map[string][]base.Z + qname string + id uuid.UUID + match func(err error) bool + wantScheduled map[string][]*base.TaskMessage + }{ + { + desc: "It should return TaskNotFoundError if task doesn't exist the queue", + scheduled: map[string][]base.Z{ + "default": {{Message: m1, Score: t1.Unix()}}, + }, + qname: "default", + id: uuid.New(), + match: errors.IsTaskNotFound, + wantScheduled: map[string][]*base.TaskMessage{ + "default": {m1}, + }, + }, + { + desc: "It should return QueueNotFoundError if the queue doesn't exist", + scheduled: map[string][]base.Z{ + "default": {{Message: m1, Score: t1.Unix()}}, + }, + qname: "nonexistent", + id: uuid.New(), + match: errors.IsQueueNotFound, + wantScheduled: map[string][]*base.TaskMessage{ + "default": {m1}, + }, + }, + } + + for _, tc := range tests { + h.FlushDB(t, r.client) // clean up db before each test case + h.SeedAllScheduledQueues(t, r.client, tc.scheduled) + + got := r.DeleteTask(tc.qname, tc.id) + if !tc.match(got) { + t.Errorf("%s: r.DeleteTask(qname, id) returned %v", tc.desc, got) + continue + } + + for qname, want := range tc.wantScheduled { + gotScheduled := h.GetScheduledMessages(t, r.client, qname) + if diff := cmp.Diff(want, gotScheduled, h.SortMsgOpt); diff != "" { + t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.ScheduledKey(qname), diff) + } + } + } +} + func TestDeleteAllArchivedTasks(t *testing.T) { r := setup(t) defer r.Close()