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

Allow task deletion by queue name and task ID

Browse Source
This commit is contained in:
Ken Hibino
2021-03-26 10:43:10 -07:00
parent 6d35d46461
commit 9bc80c6216
6 changed files with 125 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files

78
internal/rdb/inspect.go
View File

@@ -738,65 +738,37 @@ func (r *RDB) removeAndArchiveAll(src, dst 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>
// KEYS[1] -> asynq:{<qname>}:t:<task_id>
// ARGV[1] -> task ID
// ARGV[2] -> redis key prefix (asynq:{<qname>}:)
var deleteTaskCmd = redis.NewScript(`
if redis.call("ZREM", KEYS[1], ARGV[1]) == 0 then
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")
local n
if state == "PENDING" then
n = redis.call("LREM", (ARGV[2] .. "pending"), 0, ARGV[1])
elseif state == "SCHEDULED" then
n = redis.call("ZREM", (ARGV[2] .. "scheduled"), ARGV[1])
elseif state == "RETRY" then
n = redis.call("ZREM", (ARGV[2] .. "retry"), ARGV[1])
elseif state == "ARCHIVED" then
n = redis.call("ZREM", (ARGV[2] .. "archived"), ARGV[1])
else
return redis.error_reply("unknown task state: " .. tostring(state))
end
if n == 0 then
return 0
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 deletes a task that matches the given id from the given queue.
// If a task that matches the id does not exist, it returns ErrTaskNotFound.
func (r *RDB) DeleteTask(qname, id string) error {
keys := []string{base.TaskKey(qname, id)}
argv := []interface{}{id, base.QueueKeyPrefix(qname)}
res, err := deleteTaskCmd.Run(r.client, keys, argv...).Result()
if err != nil {
return err

16
internal/rdb/inspect_test.go
View File

@@ -2472,9 +2472,9 @@ func TestDeleteArchivedTask(t *testing.T) {
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)
got := r.DeleteTask(tc.qname, tc.id.String())
if got != tc.want {
t.Errorf("r.DeleteArchivedTask(%q, %v) = %v, want %v", tc.qname, tc.id, got, tc.want)
t.Errorf("r.DeleteTask(%q, %v) = %v, want %v", tc.qname, tc.id, got, tc.want)
continue
}
@@ -2553,9 +2553,9 @@ func TestDeleteRetryTask(t *testing.T) {
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)
got := r.DeleteTask(tc.qname, tc.id.String())
if got != tc.want {
t.Errorf("r.DeleteRetryTask(%q, %v) = %v, want %v", tc.qname, tc.id, got, tc.want)
t.Errorf("r.DeleteTask(%q, %v) = %v, want %v", tc.qname, tc.id, got, tc.want)
continue
}
@@ -2634,9 +2634,9 @@ func TestDeleteScheduledTask(t *testing.T) {
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)
got := r.DeleteTask(tc.qname, tc.id.String())
if got != tc.want {
t.Errorf("r.DeleteScheduledTask(%q, %v) = %v, want %v", tc.qname, tc.id, got, tc.want)
t.Errorf("r.DeleteTask(%q, %v) = %v, want %v", tc.qname, tc.id, got, tc.want)
continue
}
@@ -2704,9 +2704,9 @@ func TestDeletePendingTask(t *testing.T) {
h.FlushDB(t, r.client)
h.SeedAllPendingQueues(t, r.client, tc.pending)
got := r.DeletePendingTask(tc.qname, tc.id)
got := r.DeleteTask(tc.qname, tc.id.String())
if got != tc.want {
t.Errorf("r.DeletePendingTask(%q, %v) = %v, want %v", tc.qname, tc.id, got, tc.want)
t.Errorf("r.DeleteTask(%q, %v) = %v, want %v", tc.qname, tc.id, got, tc.want)
continue
}