WIP: Archive task by qname and id

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
 }

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:{<qname>}
-// KEYS[2] -> asynq:{<qname>}: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:{<qname>}:t:<task_id>
 // KEYS[2] -> asynq:{<qname>}:archived
-// ARGV[1] -> id of the task to archive
+// ARGV[1] -> task ID
-// ARGV[2] -> current timestamp
+// ARGV[2] -> redis key prefix (asynq:{<qname>}:)
-// ARGV[3] -> cutoff timestamp (e.g., 90 days ago)
+// ARGV[3] -> current timestamp in unix time
-// ARGV[4] -> max number of tasks in archived state (e.g., 100)
+// ARGV[4] -> cutoff timestamp (e.g., 90 days ago)
-var removeAndArchiveCmd = redis.NewScript(`
+// ARGV[5] -> max number of tasks in archived state (e.g., 100)
-if redis.call("ZREM", KEYS[1], ARGV[1]) == 0 then
+var archiveTaskCmd = redis.NewScript(`
+if redis.call("EXISTS", KEYS[1]) == 0 then
 	return 0
 end
-redis.call("ZADD", KEYS[2], ARGV[2], ARGV[1])
+local state = redis.call("HGET", KEYS[1], "state")
-redis.call("ZREMRANGEBYSCORE", KEYS[2], "-inf", ARGV[3])
+local n
-redis.call("ZREMRANGEBYRANK", KEYS[2], 0, -ARGV[4])
+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
+	argv := []interface{}{
-	res, err := removeAndArchiveCmd.Run(r.client,
+		id,
-		[]string{src, dst},
+		base.QueueKeyPrefix(qname),
-		id, now.Unix(), limit, maxArchiveSize).Result()
+		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:{<qname>}: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

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
 		}