Extract rdb to internal package

internal/rdb/rdb.go

@@ -0,0 +1,388 @@
+package rdb
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/go-redis/redis/v7"
+	"github.com/google/uuid"
+)
+
+// Redis keys
+const (
+	allQueues    = "asynq:queues"          // SET
+	queuePrefix  = "asynq:queues:"         // LIST - asynq:queues:<qname>
+	DefaultQueue = queuePrefix + "default" // LIST
+	Scheduled    = "asynq:scheduled"       // ZSET
+	Retry        = "asynq:retry"           // ZSET
+	Dead         = "asynq:dead"            // ZSET
+	InProgress   = "asynq:in_progress"     // SET
+)
+
+var ErrDequeueTimeout = errors.New("blocking dequeue operation timed out")
+
+// RDB encapsulates the interactions with redis server.
+type RDB struct {
+	client *redis.Client
+}
+
+// NewRDB returns a new instance of RDB.
+func NewRDB(client *redis.Client) *RDB {
+	return &RDB{client}
+}
+
+// TaskMessage is an internal representation of a task with additional metadata fields.
+// This data gets written in redis.
+type TaskMessage struct {
+	//-------- Task fields --------
+
+	Type    string
+	Payload map[string]interface{}
+
+	//-------- metadata fields --------
+
+	// unique identifier for each task
+	ID uuid.UUID
+
+	// queue name this message should be enqueued to
+	Queue string
+
+	// max number of retry for this task.
+	Retry int
+
+	// number of times we've retried so far
+	Retried int
+
+	// error message from the last failure
+	ErrorMsg string
+}
+
+// Stats represents a state of queues at a certain time.
+type Stats struct {
+	Queued     int
+	InProgress int
+	Scheduled  int
+	Retry      int
+	Dead       int
+	Timestamp  time.Time
+}
+
+// EnqueuedTask is a task in a queue and is ready to be processed.
+// This is read only and used for inspection purpose.
+type EnqueuedTask struct {
+	ID      uuid.UUID
+	Type    string
+	Payload map[string]interface{}
+}
+
+// InProgressTask is a task that's currently being processed.
+// This is read only and used for inspection purpose.
+type InProgressTask struct {
+	ID      uuid.UUID
+	Type    string
+	Payload map[string]interface{}
+}
+
+// ScheduledTask is a task that's scheduled to be processed in the future.
+// This is read only and used for inspection purpose.
+type ScheduledTask struct {
+	ID        uuid.UUID
+	Type      string
+	Payload   map[string]interface{}
+	ProcessAt time.Time
+}
+
+// RetryTask is a task that's in retry queue because worker failed to process the task.
+// This is read only and used for inspection purpose.
+type RetryTask struct {
+	ID      uuid.UUID
+	Type    string
+	Payload map[string]interface{}
+	// TODO(hibiken): add LastFailedAt time.Time
+	ProcessAt time.Time
+	ErrorMsg  string
+	Retried   int
+	Retry     int
+}
+
+// DeadTask is a task in that has exhausted all retries.
+// This is read only and used for inspection purpose.
+type DeadTask struct {
+	ID           uuid.UUID
+	Type         string
+	Payload      map[string]interface{}
+	LastFailedAt time.Time
+	ErrorMsg     string
+}
+
+// Close closes the connection with redis server.
+func (r *RDB) Close() error {
+	return r.client.Close()
+}
+
+// Enqueue inserts the given task to the end of the queue.
+// It also adds the queue name to the "all-queues" list.
+func (r *RDB) Enqueue(msg *TaskMessage) error {
+	bytes, err := json.Marshal(msg)
+	if err != nil {
+		return fmt.Errorf("could not marshal %+v to json: %v", msg, err)
+	}
+	qname := queuePrefix + msg.Queue
+	pipe := r.client.Pipeline()
+	pipe.SAdd(allQueues, qname)
+	pipe.LPush(qname, string(bytes))
+	_, err = pipe.Exec()
+	if err != nil {
+		return fmt.Errorf("could not enqueue the task %+v to %q: %v", msg, qname, err)
+	}
+	return nil
+}
+
+// Dequeue blocks until there is a task available to be processed,
+// once a task is available, it adds the task to "in progress" list
+// and returns the task.
+func (r *RDB) Dequeue(qname string, timeout time.Duration) (*TaskMessage, error) {
+	data, err := r.client.BRPopLPush(qname, InProgress, timeout).Result()
+	if err == redis.Nil {
+		return nil, ErrDequeueTimeout
+	}
+	if err != nil {
+		return nil, fmt.Errorf("command `BRPOPLPUSH %q %q %v` failed: %v", qname, InProgress, timeout, err)
+	}
+	var msg TaskMessage
+	err = json.Unmarshal([]byte(data), &msg)
+	if err != nil {
+		return nil, fmt.Errorf("could not unmarshal %v to json: %v", data, err)
+	}
+	fmt.Printf("[DEBUG] perform task %+v from %s\n", msg, qname)
+	return &msg, nil
+}
+
+// Remove deletes all elements equal to msg from a redis list with the given key.
+func (r *RDB) Remove(key string, msg *TaskMessage) error {
+	bytes, err := json.Marshal(msg)
+	if err != nil {
+		return fmt.Errorf("could not marshal %+v to json: %v", msg, err)
+	}
+	// NOTE: count ZERO means "remove all elements equal to val"
+	err = r.client.LRem(key, 0, string(bytes)).Err()
+	if err != nil {
+		return fmt.Errorf("command `LREM %s 0 %s` failed: %v", key, string(bytes), err)
+	}
+	return nil
+}
+
+// Schedule adds the task to the zset to be processd at the specified time.
+func (r *RDB) Schedule(zset string, processAt time.Time, msg *TaskMessage) error {
+	bytes, err := json.Marshal(msg)
+	if err != nil {
+		return fmt.Errorf("could not marshal %+v to json: %v", msg, err)
+	}
+	score := float64(processAt.Unix())
+	err = r.client.ZAdd(zset, &redis.Z{Member: string(bytes), Score: score}).Err()
+	if err != nil {
+		return fmt.Errorf("command `ZADD %s %.1f %s` failed: %v", zset, score, string(bytes), err)
+	}
+	return nil
+}
+
+const maxDeadTask = 100
+const deadExpirationInDays = 90
+
+// Kill sends the taskMessage to "dead" set.
+// It also trims the sorted set by timestamp and set size.
+func (r *RDB) Kill(msg *TaskMessage) error {
+	bytes, err := json.Marshal(msg)
+	if err != nil {
+		return fmt.Errorf("could not marshal %+v to json: %v", msg, err)
+	}
+	now := time.Now()
+	pipe := r.client.Pipeline()
+	pipe.ZAdd(Dead, &redis.Z{Member: string(bytes), Score: float64(now.Unix())})
+	limit := now.AddDate(0, 0, -deadExpirationInDays).Unix() // 90 days ago
+	pipe.ZRemRangeByScore(Dead, "-inf", strconv.Itoa(int(limit)))
+	pipe.ZRemRangeByRank(Dead, 0, -maxDeadTask) // trim the set to 100
+	_, err = pipe.Exec()
+	return err
+}
+
+// MoveAll moves all tasks from src list to dst list.
+func (r *RDB) MoveAll(src, dst string) error {
+	script := redis.NewScript(`
+	local len = redis.call("LLEN", KEYS[1])
+	for i = len, 1, -1 do
+		redis.call("RPOPLPUSH", KEYS[1], KEYS[2])
+	end
+	return len
+	`)
+	_, err := script.Run(r.client, []string{src, dst}).Result()
+	return err
+}
+
+// Forward moves all tasks with a score less than the current unix time
+// from the given zset to the default queue.
+// TODO(hibiken): Find a better method name that reflects what this does.
+func (r *RDB) Forward(from string) error {
+	script := redis.NewScript(`
+	local msgs = redis.call("ZRANGEBYSCORE", KEYS[1], "-inf", ARGV[1])
+	for _, msg in ipairs(msgs) do
+		redis.call("ZREM", KEYS[1], msg)
+		redis.call("SADD", KEYS[2], KEYS[3])
+		redis.call("LPUSH", KEYS[3], msg)
+	end
+	return msgs
+	`)
+	now := float64(time.Now().Unix())
+	res, err := script.Run(r.client, []string{from, allQueues, DefaultQueue}, now).Result()
+	fmt.Printf("[DEBUG] got %d tasks from %q\n", len(res.([]interface{})), from)
+	return err
+}
+
+// CurrentStats returns a current state of the queues.
+func (r *RDB) CurrentStats() (*Stats, error) {
+	pipe := r.client.Pipeline()
+	qlen := pipe.LLen(DefaultQueue)
+	plen := pipe.LLen(InProgress)
+	slen := pipe.ZCard(Scheduled)
+	rlen := pipe.ZCard(Retry)
+	dlen := pipe.ZCard(Dead)
+	_, err := pipe.Exec()
+	if err != nil {
+		return nil, err
+	}
+	return &Stats{
+		Queued:     int(qlen.Val()),
+		InProgress: int(plen.Val()),
+		Scheduled:  int(slen.Val()),
+		Retry:      int(rlen.Val()),
+		Dead:       int(dlen.Val()),
+		Timestamp:  time.Now(),
+	}, nil
+}
+
+func (r *RDB) ListEnqueued() ([]*TaskMessage, error) {
+	return r.rangeList(DefaultQueue)
+}
+
+func (r *RDB) ListInProgress() ([]*TaskMessage, error) {
+	return r.rangeList(InProgress)
+}
+
+func (r *RDB) ListScheduled() ([]*ScheduledTask, error) {
+	data, err := r.client.ZRangeWithScores(Scheduled, 0, -1).Result()
+	if err != nil {
+		return nil, err
+	}
+	var tasks []*ScheduledTask
+	for _, z := range data {
+		s, ok := z.Member.(string)
+		if !ok {
+			continue // bad data, ignore and continue
+		}
+		var msg TaskMessage
+		err := json.Unmarshal([]byte(s), &msg)
+		if err != nil {
+			continue // bad data, ignore and continue
+		}
+		processAt := time.Unix(int64(z.Score), 0)
+		tasks = append(tasks, &ScheduledTask{
+			ID:        msg.ID,
+			Type:      msg.Type,
+			Payload:   msg.Payload,
+			ProcessAt: processAt,
+		})
+	}
+	return tasks, nil
+}
+
+func (r *RDB) ListRetry() ([]*RetryTask, error) {
+	data, err := r.client.ZRangeWithScores(Retry, 0, -1).Result()
+	if err != nil {
+		return nil, err
+	}
+	var tasks []*RetryTask
+	for _, z := range data {
+		s, ok := z.Member.(string)
+		if !ok {
+			continue // bad data, ignore and continue
+		}
+		var msg TaskMessage
+		err := json.Unmarshal([]byte(s), &msg)
+		if err != nil {
+			continue // bad data, ignore and continue
+		}
+		processAt := time.Unix(int64(z.Score), 0)
+		tasks = append(tasks, &RetryTask{
+			ID:        msg.ID,
+			Type:      msg.Type,
+			Payload:   msg.Payload,
+			ErrorMsg:  msg.ErrorMsg,
+			Retry:     msg.Retry,
+			Retried:   msg.Retried,
+			ProcessAt: processAt,
+		})
+	}
+	return tasks, nil
+}
+
+func (r *RDB) ListDead() ([]*DeadTask, error) {
+	data, err := r.client.ZRangeWithScores(Dead, 0, -1).Result()
+	if err != nil {
+		return nil, err
+	}
+	var tasks []*DeadTask
+	for _, z := range data {
+		s, ok := z.Member.(string)
+		if !ok {
+			continue // bad data, ignore and continue
+		}
+		var msg TaskMessage
+		err := json.Unmarshal([]byte(s), &msg)
+		if err != nil {
+			continue // bad data, ignore and continue
+		}
+		lastFailedAt := time.Unix(int64(z.Score), 0)
+		tasks = append(tasks, &DeadTask{
+			ID:           msg.ID,
+			Type:         msg.Type,
+			Payload:      msg.Payload,
+			ErrorMsg:     msg.ErrorMsg,
+			LastFailedAt: lastFailedAt,
+		})
+	}
+	return tasks, nil
+}
+
+func (r *RDB) rangeList(key string) ([]*TaskMessage, error) {
+	data, err := r.client.LRange(key, 0, -1).Result()
+	if err != nil {
+		return nil, err
+	}
+	return r.toMessageSlice(data), nil
+}
+
+func (r *RDB) rangeZSet(key string) ([]*TaskMessage, error) {
+	data, err := r.client.ZRange(key, 0, -1).Result()
+	if err != nil {
+		return nil, err
+	}
+	return r.toMessageSlice(data), nil
+}
+
+// toMessageSlice convers json strings to a slice of task messages.
+func (r *RDB) toMessageSlice(data []string) []*TaskMessage {
+	var msgs []*TaskMessage
+	for _, s := range data {
+		var msg TaskMessage
+		err := json.Unmarshal([]byte(s), &msg)
+		if err != nil {
+			// bad data; ignore and continue
+			continue
+		}
+		msgs = append(msgs, &msg)
+	}
+	return msgs
+}

internal/rdb/rdb_test.go

@@ -0,0 +1,432 @@
+package rdb
+
+import (
+	"encoding/json"
+	"fmt"
+	"math/rand"
+	"sort"
+	"testing"
+	"time"
+
+	"github.com/go-redis/redis/v7"
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/uuid"
+)
+
+func init() {
+	rand.Seed(time.Now().UnixNano())
+}
+
+func setup(t *testing.T) *RDB {
+	t.Helper()
+	r := NewRDB(redis.NewClient(&redis.Options{
+		Addr: "localhost:6379",
+		DB:   15,
+	}))
+	// Start each test with a clean slate.
+	if err := r.client.FlushDB().Err(); err != nil {
+		panic(err)
+	}
+	return r
+}
+
+var sortMsgOpt = cmp.Transformer("SortMsg", func(in []*TaskMessage) []*TaskMessage {
+	out := append([]*TaskMessage(nil), in...) // Copy input to avoid mutating it
+	sort.Slice(out, func(i, j int) bool {
+		return out[i].ID.String() < out[j].ID.String()
+	})
+	return out
+})
+
+func randomTask(taskType, qname string, payload map[string]interface{}) *TaskMessage {
+	return &TaskMessage{
+		ID:      uuid.New(),
+		Type:    taskType,
+		Queue:   qname,
+		Retry:   25,
+		Payload: make(map[string]interface{}),
+	}
+}
+
+func mustMarshal(t *testing.T, task *TaskMessage) string {
+	t.Helper()
+	data, err := json.Marshal(task)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return string(data)
+}
+
+func mustUnmarshal(t *testing.T, data string) *TaskMessage {
+	t.Helper()
+	var task TaskMessage
+	err := json.Unmarshal([]byte(data), &task)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return &task
+}
+
+func mustMarshalSlice(t *testing.T, tasks []*TaskMessage) []string {
+	t.Helper()
+	var data []string
+	for _, task := range tasks {
+		data = append(data, mustMarshal(t, task))
+	}
+	return data
+}
+
+func mustUnmarshalSlice(t *testing.T, data []string) []*TaskMessage {
+	t.Helper()
+	var tasks []*TaskMessage
+	for _, s := range data {
+		tasks = append(tasks, mustUnmarshal(t, s))
+	}
+	return tasks
+}
+
+func TestEnqueue(t *testing.T) {
+	r := setup(t)
+	tests := []struct {
+		msg *TaskMessage
+	}{
+		{msg: randomTask("send_email", "default",
+			map[string]interface{}{"to": "exampleuser@gmail.com", "from": "noreply@example.com"})},
+		{msg: randomTask("generate_csv", "default",
+			map[string]interface{}{})},
+		{msg: randomTask("sync", "default", nil)},
+	}
+
+	for _, tc := range tests {
+		// clean up db before each test case.
+		if err := r.client.FlushDB().Err(); err != nil {
+			t.Fatal(err)
+		}
+		err := r.Enqueue(tc.msg)
+		if err != nil {
+			t.Error(err)
+			continue
+		}
+		res := r.client.LRange(DefaultQueue, 0, -1).Val()
+		if len(res) != 1 {
+			t.Errorf("LIST %q has length %d, want 1", DefaultQueue, len(res))
+			continue
+		}
+		if !r.client.SIsMember(allQueues, DefaultQueue).Val() {
+			t.Errorf("SISMEMBER %q %q = false, want true", allQueues, DefaultQueue)
+		}
+		if diff := cmp.Diff(*tc.msg, *mustUnmarshal(t, res[0])); diff != "" {
+			t.Errorf("persisted data differed from the original input (-want, +got)\n%s", diff)
+		}
+	}
+}
+
+func TestDequeue(t *testing.T) {
+	r := setup(t)
+	t1 := randomTask("send_email", "default", map[string]interface{}{"subject": "hello!"})
+	tests := []struct {
+		queued     []*TaskMessage
+		want       *TaskMessage
+		err        error
+		inProgress int64 // length of "in-progress" tasks after dequeue
+	}{
+		{queued: []*TaskMessage{t1}, want: t1, err: nil, inProgress: 1},
+		{queued: []*TaskMessage{}, want: nil, err: ErrDequeueTimeout, inProgress: 0},
+	}
+
+	for _, tc := range tests {
+		// clean up db before each test case.
+		if err := r.client.FlushDB().Err(); err != nil {
+			t.Fatal(err)
+		}
+		for _, m := range tc.queued {
+			r.Enqueue(m)
+		}
+		got, err := r.Dequeue(DefaultQueue, time.Second)
+		if !cmp.Equal(got, tc.want) || err != tc.err {
+			t.Errorf("(*rdb).dequeue(%q, time.Second) = %v, %v; want %v, %v",
+				DefaultQueue, got, err, tc.want, tc.err)
+			continue
+		}
+		if l := r.client.LLen(InProgress).Val(); l != tc.inProgress {
+			t.Errorf("LIST %q has length %d, want %d", InProgress, l, tc.inProgress)
+		}
+	}
+}
+
+func TestRemove(t *testing.T) {
+	r := setup(t)
+	t1 := randomTask("send_email", "default", nil)
+	t2 := randomTask("export_csv", "csv", nil)
+
+	tests := []struct {
+		initial []*TaskMessage // initial state of the list
+		target  *TaskMessage   // task to remove
+		final   []*TaskMessage // final state of the list
+	}{
+		{
+			initial: []*TaskMessage{t1, t2},
+			target:  t1,
+			final:   []*TaskMessage{t2},
+		},
+		{
+			initial: []*TaskMessage{t2},
+			target:  t1,
+			final:   []*TaskMessage{t2},
+		},
+		{
+			initial: []*TaskMessage{t1},
+			target:  t1,
+			final:   []*TaskMessage{},
+		},
+	}
+
+	for _, tc := range tests {
+		// clean up db before each test case.
+		if err := r.client.FlushDB().Err(); err != nil {
+			t.Fatal(err)
+		}
+		// set up initial state
+		for _, task := range tc.initial {
+			err := r.client.LPush(DefaultQueue, mustMarshal(t, task)).Err()
+			if err != nil {
+				t.Fatal(err)
+			}
+		}
+
+		err := r.Remove(DefaultQueue, tc.target)
+		if err != nil {
+			t.Error(err)
+			continue
+		}
+
+		var got []*TaskMessage
+		data := r.client.LRange(DefaultQueue, 0, -1).Val()
+		for _, s := range data {
+			got = append(got, mustUnmarshal(t, s))
+		}
+
+		if diff := cmp.Diff(tc.final, got, sortMsgOpt); diff != "" {
+			t.Errorf("mismatch found in %q after calling (*rdb).remove: (-want, +got):\n%s", DefaultQueue, diff)
+			continue
+		}
+	}
+}
+
+func TestKill(t *testing.T) {
+	r := setup(t)
+	t1 := randomTask("send_email", "default", nil)
+
+	// TODO(hibiken): add test cases for trimming
+	tests := []struct {
+		initial []*TaskMessage //  inital state of "dead" set
+		target  *TaskMessage   // task to kill
+		want    []*TaskMessage // final state of "dead" set
+	}{
+		{
+			initial: []*TaskMessage{},
+			target:  t1,
+			want:    []*TaskMessage{t1},
+		},
+	}
+
+	for _, tc := range tests {
+		// clean up db before each test case.
+		if err := r.client.FlushDB().Err(); err != nil {
+			t.Fatal(err)
+		}
+		// set up initial state
+		for _, task := range tc.initial {
+			err := r.client.ZAdd(Dead, &redis.Z{Member: mustMarshal(t, task), Score: float64(time.Now().Unix())}).Err()
+			if err != nil {
+				t.Fatal(err)
+			}
+		}
+
+		err := r.Kill(tc.target)
+		if err != nil {
+			t.Error(err)
+			continue
+		}
+
+		actual := r.client.ZRange(Dead, 0, -1).Val()
+		got := mustUnmarshalSlice(t, actual)
+		if diff := cmp.Diff(tc.want, got, sortMsgOpt); diff != "" {
+			t.Errorf("mismatch found in %q after calling (*rdb).kill: (-want, +got):\n%s", Dead, diff)
+			continue
+		}
+	}
+}
+
+func TestMoveAll(t *testing.T) {
+	r := setup(t)
+	t1 := randomTask("send_email", "default", nil)
+	t2 := randomTask("export_csv", "csv", nil)
+	t3 := randomTask("sync_stuff", "sync", nil)
+
+	tests := []struct {
+		beforeSrc []*TaskMessage
+		beforeDst []*TaskMessage
+		afterSrc  []*TaskMessage
+		afterDst  []*TaskMessage
+	}{
+		{
+			beforeSrc: []*TaskMessage{t1, t2, t3},
+			beforeDst: []*TaskMessage{},
+			afterSrc:  []*TaskMessage{},
+			afterDst:  []*TaskMessage{t1, t2, t3},
+		},
+		{
+			beforeSrc: []*TaskMessage{},
+			beforeDst: []*TaskMessage{t1, t2, t3},
+			afterSrc:  []*TaskMessage{},
+			afterDst:  []*TaskMessage{t1, t2, t3},
+		},
+		{
+			beforeSrc: []*TaskMessage{t2, t3},
+			beforeDst: []*TaskMessage{t1},
+			afterSrc:  []*TaskMessage{},
+			afterDst:  []*TaskMessage{t1, t2, t3},
+		},
+	}
+
+	for _, tc := range tests {
+		// clean up db before each test case.
+		if err := r.client.FlushDB().Err(); err != nil {
+			t.Error(err)
+			continue
+		}
+		// seed src list.
+		for _, msg := range tc.beforeSrc {
+			r.client.LPush(InProgress, mustMarshal(t, msg))
+		}
+		// seed dst list.
+		for _, msg := range tc.beforeDst {
+			r.client.LPush(DefaultQueue, mustMarshal(t, msg))
+		}
+
+		if err := r.MoveAll(InProgress, DefaultQueue); err != nil {
+			t.Errorf("(*rdb).moveAll(%q, %q) = %v, want nil", InProgress, DefaultQueue, err)
+			continue
+		}
+
+		src := r.client.LRange(InProgress, 0, -1).Val()
+		gotSrc := mustUnmarshalSlice(t, src)
+		if diff := cmp.Diff(tc.afterSrc, gotSrc, sortMsgOpt); diff != "" {
+			t.Errorf("mismatch found in %q (-want, +got)\n%s", InProgress, diff)
+		}
+		dst := r.client.LRange(DefaultQueue, 0, -1).Val()
+		gotDst := mustUnmarshalSlice(t, dst)
+		if diff := cmp.Diff(tc.afterDst, gotDst, sortMsgOpt); diff != "" {
+			t.Errorf("mismatch found in %q (-want, +got)\n%s", DefaultQueue, diff)
+		}
+	}
+}
+
+func TestForward(t *testing.T) {
+	r := setup(t)
+	t1 := randomTask("send_email", "default", nil)
+	t2 := randomTask("generate_csv", "default", nil)
+	secondAgo := time.Now().Add(-time.Second)
+	hourFromNow := time.Now().Add(time.Hour)
+
+	tests := []struct {
+		tasks         []*redis.Z     // scheduled tasks with timestamp as a score
+		wantQueued    []*TaskMessage // queue after calling forward
+		wantScheduled []*TaskMessage // scheduled queue after calling forward
+	}{
+		{
+			tasks: []*redis.Z{
+				&redis.Z{Member: mustMarshal(t, t1), Score: float64(secondAgo.Unix())},
+				&redis.Z{Member: mustMarshal(t, t2), Score: float64(secondAgo.Unix())}},
+			wantQueued:    []*TaskMessage{t1, t2},
+			wantScheduled: []*TaskMessage{},
+		},
+		{
+			tasks: []*redis.Z{
+				&redis.Z{Member: mustMarshal(t, t1), Score: float64(hourFromNow.Unix())},
+				&redis.Z{Member: mustMarshal(t, t2), Score: float64(secondAgo.Unix())}},
+			wantQueued:    []*TaskMessage{t2},
+			wantScheduled: []*TaskMessage{t1},
+		},
+		{
+			tasks: []*redis.Z{
+				&redis.Z{Member: mustMarshal(t, t1), Score: float64(hourFromNow.Unix())},
+				&redis.Z{Member: mustMarshal(t, t2), Score: float64(hourFromNow.Unix())}},
+			wantQueued:    []*TaskMessage{},
+			wantScheduled: []*TaskMessage{t1, t2},
+		},
+	}
+
+	for _, tc := range tests {
+		// clean up db before each test case.
+		if err := r.client.FlushDB().Err(); err != nil {
+			t.Fatal(err)
+		}
+		if err := r.client.ZAdd(Scheduled, tc.tasks...).Err(); err != nil {
+			t.Error(err)
+			continue
+		}
+
+		err := r.Forward(Scheduled)
+		if err != nil {
+			t.Errorf("(*rdb).forward(%q) = %v, want nil", Scheduled, err)
+			continue
+		}
+		queued := r.client.LRange(DefaultQueue, 0, -1).Val()
+		gotQueued := mustUnmarshalSlice(t, queued)
+		if diff := cmp.Diff(tc.wantQueued, gotQueued, sortMsgOpt); diff != "" {
+			t.Errorf("%q has %d tasks, want %d tasks; (-want, +got)\n%s", DefaultQueue, len(gotQueued), len(tc.wantQueued), diff)
+		}
+		scheduled := r.client.ZRangeByScore(Scheduled, &redis.ZRangeBy{Min: "-inf", Max: "+inf"}).Val()
+		gotScheduled := mustUnmarshalSlice(t, scheduled)
+		if diff := cmp.Diff(tc.wantScheduled, gotScheduled, sortMsgOpt); diff != "" {
+			t.Errorf("%q has %d tasks, want %d tasks; (-want, +got)\n%s", scheduled, len(gotScheduled), len(tc.wantScheduled), diff)
+		}
+	}
+}
+
+func TestSchedule(t *testing.T) {
+	r := setup(t)
+	tests := []struct {
+		msg       *TaskMessage
+		processAt time.Time
+		zset      string
+	}{
+		{
+			randomTask("send_email", "default", map[string]interface{}{"subject": "hello"}),
+			time.Now().Add(15 * time.Minute),
+			Scheduled,
+		},
+	}
+
+	for _, tc := range tests {
+		// clean up db before each test case.
+		if err := r.client.FlushDB().Err(); err != nil {
+			t.Fatal(err)
+		}
+
+		err := r.Schedule(tc.zset, tc.processAt, tc.msg)
+		if err != nil {
+			t.Error(err)
+			continue
+		}
+
+		res, err := r.client.ZRangeWithScores(tc.zset, 0, -1).Result()
+		if err != nil {
+			t.Error(err)
+			continue
+		}
+
+		desc := fmt.Sprintf("(*rdb).schedule(%q, %v, %v)", tc.zset, tc.processAt, tc.msg)
+		if len(res) != 1 {
+			t.Errorf("%s inserted %d items to %q, want 1 items inserted", desc, len(res), tc.zset)
+			continue
+		}
+
+		if res[0].Score != float64(tc.processAt.Unix()) {
+			t.Errorf("%s inserted an item with score %f, want %f", desc, res[0].Score, float64(tc.processAt.Unix()))
+			continue
+		}
+	}
+}