mirror of
https://github.com/hibiken/asynq.git
synced 2024-11-10 11:31:58 +08:00
Update RDB.DeleteTask with task state
This commit is contained in:
Ken Hibino
parent
675826be5f
commit
430dbb298e
@ -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:{<qname>}:pending
|
||||
// KEYS[2] -> asynq:{<qname>}:t:<task_id>
|
||||
// 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:{<qname>}:retry)
|
||||
// KEYS[2] -> asynq:{<qname>}:t:<task_id>
|
||||
// Input:
|
||||
// KEYS[1] -> asynq:{<qname>}:t:<task_id>
|
||||
// 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
|
||||
|
||||
@ -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()
|
||||
|
||||
Loading…
Reference in New Issue
Block a user