From 4bb67c582d572d7f1c92f687cacd6b6bd751476c Mon Sep 17 00:00:00 2001 From: Ken Hibino Date: Fri, 26 Mar 2021 11:47:07 -0700 Subject: [PATCH] WIP: Archive task by qname and id --- inspeq/inspector.go | 13 +++- internal/rdb/inspect.go | 141 +++++++++++++---------------------- internal/rdb/inspect_test.go | 14 ++-- 3 files changed, 69 insertions(+), 99 deletions(-) diff --git a/inspeq/inspector.go b/inspeq/inspector.go index 11c4290..c7c8f46 100644 --- a/inspeq/inspector.go +++ b/inspeq/inspector.go @@ -179,29 +179,40 @@ func (i *Inspector) DeleteQueue(qname string, force bool) error { type TaskInfo interface { // ID returns the unique identifier of the task. ID() string + // Type returns the type name of the task. Type() string + // Payload returns the payload data of the task. Payload() []byte + // State returns a TaskState representing the state of the task. State() TaskState + // Queue returns the name of the queue the task belongs to. Queue() string + // MaxRetry returns the maximum number of times the task can be retried. MaxRetry() int + // Retried returns the number of times the task has been retried. Retried() int + // Deadline returns the deadline set for the task. Deadline() time.Time + // Timeout returns the timeout duration set for the task. Timeout() time.Duration + // NextProcessAt returns the time the task will be ready to be processed. // Zero value of time.Time is used when task is in pending, active, or archived // state. NextProcessAt() time.Time + // LastFailedAt returns the time the task last failed. // Zero value of time.Time is used if the task has not failed. LastFailedAt() time.Time + // LastErr returns the error message from the last failure. // Empty string is returned if the task has not failed. LastErr() string @@ -218,7 +229,7 @@ const ( TaskStateArchived ) -func (i *Inspector) GetTaskInfo(taskID string) (TaskInfo, error) { +func (i *Inspector) GetTaskInfo(qname, id string) (TaskInfo, error) { // TODO: implement this return nil, nil } diff --git a/internal/rdb/inspect.go b/internal/rdb/inspect.go index 2dda829..faca3f5 100644 --- a/internal/rdb/inspect.go +++ b/internal/rdb/inspect.go @@ -551,79 +551,6 @@ func (r *RDB) removeAndRunAll(zset, qkey string) (int64, error) { return n, nil } -// ArchiveRetryTask finds a retry task that matches the given id -// from the given queue and archives it. -// If there's no match, it returns ErrTaskNotFound. -func (r *RDB) ArchiveRetryTask(qname string, id uuid.UUID) error { - n, err := r.removeAndArchive(base.RetryKey(qname), base.ArchivedKey(qname), id.String()) - if err != nil { - return err - } - if n == 0 { - return ErrTaskNotFound - } - return nil -} - -// ArchiveScheduledTask finds a scheduled task that matches the given id -// from the given queue and archives it. -// If there's no match, it returns ErrTaskNotFound. -func (r *RDB) ArchiveScheduledTask(qname string, id uuid.UUID) error { - n, err := r.removeAndArchive(base.ScheduledKey(qname), base.ArchivedKey(qname), id.String()) - if err != nil { - return err - } - if n == 0 { - return ErrTaskNotFound - } - return nil -} - -// KEYS[1] -> asynq:{} -// KEYS[2] -> asynq:{}:archived -// ARGV[1] -> ID of the task to archive -// ARGV[2] -> current timestamp -// ARGV[3] -> cutoff timestamp (e.g., 90 days ago) -// ARGV[4] -> max number of tasks in archive (e.g., 100) -var archivePendingCmd = redis.NewScript(` -if redis.call("LREM", KEYS[1], 1, ARGV[1]) == 0 then - return 0 -end -redis.call("ZADD", KEYS[2], ARGV[2], ARGV[1]) -redis.call("ZREMRANGEBYSCORE", KEYS[2], "-inf", ARGV[3]) -redis.call("ZREMRANGEBYRANK", KEYS[2], 0, -ARGV[4]) -return 1 -`) - -// ArchivePendingTask finds a pending task that matches the given id -// from the given queue and archives it. -// If there's no match, it returns ErrTaskNotFound. -func (r *RDB) ArchivePendingTask(qname string, id uuid.UUID) error { - keys := []string{ - base.PendingKey(qname), - base.ArchivedKey(qname), - } - now := time.Now() - argv := []interface{}{ - id.String(), - now.Unix(), - now.AddDate(0, 0, -archivedExpirationInDays).Unix(), - maxArchiveSize, - } - res, err := archivePendingCmd.Run(r.client, keys, argv...).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 -} - // ArchiveAllRetryTasks archives all retry tasks from the given queue and // returns the number of tasks that were moved. func (r *RDB) ArchiveAllRetryTasks(qname string) (int64, error) { @@ -672,36 +599,66 @@ func (r *RDB) ArchiveAllPendingTasks(qname string) (int64, error) { return n, nil } -// KEYS[1] -> ZSET to move task from (e.g., retry queue) +// KEYS[1] -> asynq:{}:t: // KEYS[2] -> asynq:{}:archived -// ARGV[1] -> id of the task to archive -// ARGV[2] -> current timestamp -// ARGV[3] -> cutoff timestamp (e.g., 90 days ago) -// ARGV[4] -> max number of tasks in archived state (e.g., 100) -var removeAndArchiveCmd = redis.NewScript(` -if redis.call("ZREM", KEYS[1], ARGV[1]) == 0 then +// ARGV[1] -> task ID +// ARGV[2] -> redis key prefix (asynq:{}:) +// ARGV[3] -> current timestamp in unix time +// ARGV[4] -> cutoff timestamp (e.g., 90 days ago) +// ARGV[5] -> max number of tasks in archived state (e.g., 100) +var archiveTaskCmd = redis.NewScript(` +if redis.call("EXISTS", KEYS[1]) == 0 then return 0 end -redis.call("ZADD", KEYS[2], ARGV[2], ARGV[1]) -redis.call("ZREMRANGEBYSCORE", KEYS[2], "-inf", ARGV[3]) -redis.call("ZREMRANGEBYRANK", KEYS[2], 0, -ARGV[4]) +local state = redis.call("HGET", KEYS[1], "state") +local n +if state == "PENDING" then + n = redis.call("LREM", (ARGV[2] .. "pending"), 1, 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 + return redis.error_reply("task is already archived") +elseif state == "ACTIVE" then + return redis.error_reply("cannot archive active task") +else + return redis.error_reply("unknown task state: " .. tostring(state)) +end +if n == 0 then + return 0 +end +redis.call("ZADD", KEYS[2], ARGV[3], ARGV[1]) +redis.call("ZREMRANGEBYSCORE", KEYS[2], "-inf", ARGV[4]) +redis.call("ZREMRANGEBYRANK", KEYS[2], 0, -ARGV[5]) return 1 `) -func (r *RDB) removeAndArchive(src, dst, id string) (int64, error) { +func (r *RDB) ArchiveTask(qname, id string) error { + keys := []string{ + base.TaskKey(qname, id), + base.ArchivedKey(qname), + } now := time.Now() - limit := now.AddDate(0, 0, -archivedExpirationInDays).Unix() // 90 days ago - res, err := removeAndArchiveCmd.Run(r.client, - []string{src, dst}, - id, now.Unix(), limit, maxArchiveSize).Result() + argv := []interface{}{ + id, + base.QueueKeyPrefix(qname), + now.Unix(), + now.AddDate(0, 0, -archivedExpirationInDays).Unix(), + maxArchiveSize, + } + res, err := archiveTaskCmd.Run(r.client, keys, argv...).Result() if err != nil { - return 0, err + return err } n, ok := res.(int64) if !ok { - return 0, fmt.Errorf("could not cast %v to int64", res) + return fmt.Errorf("command error: unexpected return value %v", res) } - return n, nil + if n == 0 { + return ErrTaskNotFound + } + return nil } // KEYS[1] -> ZSET to move task from (e.g., asynq:{}:retry) @@ -755,6 +712,8 @@ 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]) +elseif state == "ACTIVE" + return redis.error_reply("cannot delete active task") else return redis.error_reply("unknown task state: " .. tostring(state)) end diff --git a/internal/rdb/inspect_test.go b/internal/rdb/inspect_test.go index ce326e3..48e6669 100644 --- a/internal/rdb/inspect_test.go +++ b/internal/rdb/inspect_test.go @@ -1714,9 +1714,9 @@ func TestArchiveRetryTask(t *testing.T) { h.SeedAllRetryQueues(t, r.client, tc.retry) h.SeedAllArchivedQueues(t, r.client, tc.archived) - got := r.ArchiveRetryTask(tc.qname, tc.id) + got := r.ArchiveTask(tc.qname, tc.id.String()) if got != tc.want { - t.Errorf("(*RDB).ArchiveRetryTask(%q, %v) = %v, want %v", + t.Errorf("(*RDB).ArchiveTask(%q, %v) = %v, want %v", tc.qname, tc.id, got, tc.want) continue } @@ -1836,9 +1836,9 @@ func TestArchiveScheduledTask(t *testing.T) { h.SeedAllScheduledQueues(t, r.client, tc.scheduled) h.SeedAllArchivedQueues(t, r.client, tc.archived) - got := r.ArchiveScheduledTask(tc.qname, tc.id) + got := r.ArchiveTask(tc.qname, tc.id.String()) if got != tc.want { - t.Errorf("(*RDB).ArchiveScheduledTask(%q, %v) = %v, want %v", + t.Errorf("(*RDB).ArchiveTask(%q, %v) = %v, want %v", tc.qname, tc.id, got, tc.want) continue } @@ -1906,7 +1906,7 @@ func TestArchivePendingTask(t *testing.T) { }, qname: "default", id: m2.ID, - want: ErrTaskNotFound, + want: ErrTaskNotFound, // TODO: change to appropriate error (e.g. IllegalOperationError) wantPending: map[string][]*base.TaskMessage{ "default": {m1}, }, @@ -1942,9 +1942,9 @@ func TestArchivePendingTask(t *testing.T) { h.SeedAllPendingQueues(t, r.client, tc.pending) h.SeedAllArchivedQueues(t, r.client, tc.archived) - got := r.ArchivePendingTask(tc.qname, tc.id) + got := r.ArchiveTask(tc.qname, tc.id.String()) if got != tc.want { - t.Errorf("(*RDB).ArchivePendingTask(%q, %v) = %v, want %v", + t.Errorf("(*RDB).ArchiveTask(%q, %v) = %v, want %v", tc.qname, tc.id, got, tc.want) continue }