Repository: hibiken/asynq
Pull request #364 (2021-12-11 15:07:18)

Report Queue latency
Checks the item listed in #246 ✅ 

codecov[bot]: # [Codecov](https://codecov.io/gh/hibiken/asynq/pull/364?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Ken+Hibino) Report
> Merging [#364](https://codecov.io/gh/hibiken/asynq/pull/364?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Ken+Hibino) (4a44615) into [master](https://codecov.io/gh/hibiken/asynq/commit/9f2c321e9804901f06b58e82f9e0ce4125d80b74?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Ken+Hibino) (9f2c321) will **increase** coverage by `0.10%`.
> The diff coverage is `100.00%`.

[![Impacted file tree graph](https://codecov.io/gh/hibiken/asynq/pull/364/graphs/tree.svg?width=650&height=150&src=pr&token=BAKLSWL2G3&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Ken+Hibino)](https://codecov.io/gh/hibiken/asynq/pull/364?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Ken+Hibino)

```diff
@@            Coverage Diff             @@
##           master     #364      +/-   ##
==========================================
+ Coverage   73.49%   73.60%   +0.10%     
==========================================
  Files          23       23              
  Lines        2962     2974      +12     
==========================================
+ Hits         2177     2189      +12     
  Misses        555      555              
  Partials      230      230              
```


| [Impacted Files](https://codecov.io/gh/hibiken/asynq/pull/364?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Ken+Hibino) | Coverage Δ | |
|---|---|---|
| [inspector.go](https://codecov.io/gh/hibiken/asynq/pull/364/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Ken+Hibino#diff-aW5zcGVjdG9yLmdv) | `60.68% <100.00%> (+0.10%)` | :arrow_up: |
| [internal/rdb/inspect.go](https://codecov.io/gh/hibiken/asynq/pull/364/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Ken+Hibino#diff-aW50ZXJuYWwvcmRiL2luc3BlY3QuZ28=) | `67.89% <100.00%> (+0.23%)` | :arrow_up: |
| [internal/rdb/rdb.go](https://codecov.io/gh/hibiken/asynq/pull/364/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Ken+Hibino#diff-aW50ZXJuYWwvcmRiL3JkYi5nbw==) | `72.34% <100.00%> (+0.44%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/hibiken/asynq/pull/364?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Ken+Hibino).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Ken+Hibino)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/hibiken/asynq/pull/364?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Ken+Hibino). Last update [9f2c321...4a44615](https://codecov.io/gh/hibiken/asynq/pull/364?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Ken+Hibino). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Ken+Hibino).


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index e28b9ba..9108c10 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- `Latency` field is added to `QueueInfo`.
 - `EnqueueContext` method is added to `Client`.
 
 ### Fixed
diff --git a/inspector.go b/inspector.go
index 870f36f..1b535fe 100644
--- a/inspector.go
+++ b/inspector.go
@@ -52,6 +52,9 @@ type QueueInfo struct {
 	// It is an approximate memory usage value in bytes since the value is computed by sampling.
 	MemoryUsage int64
 
+	// Latency of the queue, measured by the oldest pending task in the queue.
+	Latency time.Duration
+
 	// Size is the total number of tasks in the queue.
 	// The value is the sum of Pending, Active, Scheduled, Retry, and Archived.
 	Size int
@@ -95,6 +98,7 @@ func (i *Inspector) GetQueueInfo(qname string) (*QueueInfo, error) {
 	return &QueueInfo{
 		Queue:       stats.Queue,
 		MemoryUsage: stats.MemoryUsage,
+		Latency:     stats.Latency,
 		Size:        stats.Size,
 		Pending:     stats.Pending,
 		Active:      stats.Active,
diff --git a/internal/rdb/inspect.go b/internal/rdb/inspect.go
index c7e1eb3..0612b38 100644
--- a/internal/rdb/inspect.go
+++ b/internal/rdb/inspect.go
@@ -46,6 +46,8 @@ type Stats struct {
 	Processed int
 	// Total number of tasks failed during the current date.
 	Failed int
+	// Latency of the queue, measured by the oldest pending task in the queue.
+	Latency time.Duration
 	// Time this stats was taken.
 	Timestamp time.Time
 }
@@ -63,7 +65,7 @@ type DailyStats struct {
 	Time time.Time
 }
 
-// KEYS[1] -> asynq:<qname>
+// KEYS[1] -> asynq:<qname>:pending
 // KEYS[2] -> asynq:<qname>:active
 // KEYS[3] -> asynq:<qname>:scheduled
 // KEYS[4] -> asynq:<qname>:retry
@@ -72,10 +74,13 @@ type DailyStats struct {
 // KEYS[7] -> asynq:<qname>:processed:<yyyy-mm-dd>
 // KEYS[8] -> asynq:<qname>:failed:<yyyy-mm-dd>
 // KEYS[9] -> asynq:<qname>:paused
+//
+// ARGV[1] -> task key prefix
 var currentStatsCmd = redis.NewScript(`
 local res = {}
+local pendingTaskCount = redis.call("LLEN", KEYS[1])
 table.insert(res, KEYS[1])
-table.insert(res, redis.call("LLEN", KEYS[1]))
+table.insert(res, pendingTaskCount)
 table.insert(res, KEYS[2])
 table.insert(res, redis.call("LLEN", KEYS[2]))
 table.insert(res, KEYS[3])
@@ -102,6 +107,13 @@ table.insert(res, KEYS[8])
 table.insert(res, fcount)
 table.insert(res, KEYS[9])
 table.insert(res, redis.call("EXISTS", KEYS[9]))
+table.insert(res, "oldest_pending_since")
+if pendingTaskCount > 0 then
+	local id = redis.call("LRANGE", KEYS[1], -1, -1)[1]
+	table.insert(res, redis.call("HGET", ARGV[1] .. id, "pending_since"))
+else
+	table.insert(res, 0)
+end
 return res`)
 
 // CurrentStats returns a current state of the queues.
@@ -125,7 +137,7 @@ func (r *RDB) CurrentStats(qname string) (*Stats, error) {
 		base.ProcessedKey(qname, now),
 		base.FailedKey(qname, now),
 		base.PausedKey(qname),
-	}).Result()
+	}, base.TaskKeyPrefix(qname)).Result()
 	if err != nil {
 		return nil, errors.E(op, errors.Unknown, err)
 	}
@@ -170,6 +182,12 @@ func (r *RDB) CurrentStats(qname string) (*Stats, error) {
 			} else {
 				stats.Paused = true
 			}
+		case "oldest_pending_since":
+			if val == 0 {
+				stats.Latency = 0
+			} else {
+				stats.Latency = r.clock.Now().Sub(time.Unix(0, int64(val)))
+			}
 		}
 	}
 	stats.Size = size
diff --git a/internal/rdb/rdb.go b/internal/rdb/rdb.go
index a2a0fe7..4152618 100644
--- a/internal/rdb/rdb.go
+++ b/internal/rdb/rdb.go
@@ -13,6 +13,7 @@ import (
 	"github.com/go-redis/redis/v8"
 	"github.com/hibiken/asynq/internal/base"
 	"github.com/hibiken/asynq/internal/errors"
+	"github.com/hibiken/asynq/internal/timeutil"
 	"github.com/spf13/cast"
 )
 
@@ -21,11 +22,15 @@ const statsTTL = 90 * 24 * time.Hour // 90 days
 // RDB is a client interface to query and mutate task queues.
 type RDB struct {
 	client redis.UniversalClient
+	clock  timeutil.Clock
 }
 
 // NewRDB returns a new instance of RDB.
 func NewRDB(client redis.UniversalClient) *RDB {
-	return &RDB{client}
+	return &RDB{
+		client: client,
+		clock:  timeutil.NewRealClock(),
+	}
 }
 
 // Close closes the connection with redis server.
@@ -38,6 +43,13 @@ func (r *RDB) Client() redis.UniversalClient {
 	return r.client
 }
 
+// SetClock sets the clock used by RDB to the given clock.
+//
+// Use this function to set the clock to SimulatedClock in tests.
+func (r *RDB) SetClock(c timeutil.Clock) {
+	r.clock = c
+}
+
 // Ping checks the connection with redis server.
 func (r *RDB) Ping() error {
 	return r.client.Ping(context.Background()).Err()
@@ -73,6 +85,7 @@ func (r *RDB) runScriptWithErrorCode(ctx context.Context, op errors.Op, script *
 // ARGV[2] -> task ID
 // ARGV[3] -> task timeout in seconds (0 if not timeout)
 // ARGV[4] -> task deadline in unix time (0 if no deadline)
+// ARGV[5] -> current unix time in nsec
 //
 // Output:
 // Returns 1 if successfully enqueued
@@ -85,7 +98,8 @@ redis.call("HSET", KEYS[1],
            "msg", ARGV[1],
            "state", "pending",
            "timeout", ARGV[3],
-           "deadline", ARGV[4])
+           "deadline", ARGV[4],
+           "pending_since", ARGV[5])
 redis.call("LPUSH", KEYS[2], ARGV[2])
 return 1
 `)
@@ -109,6 +123,7 @@ func (r *RDB) Enqueue(ctx context.Context, msg *base.TaskMessage) error {
 		msg.ID,
 		msg.Timeout,
 		msg.Deadline,
+		r.clock.Now().UnixNano(),
 	}
 	n, err := r.runScriptWithErrorCode(ctx, op, enqueueCmd, keys, argv...)
 	if err != nil {
@@ -131,6 +146,7 @@ func (r *RDB) Enqueue(ctx context.Context, msg *base.TaskMessage) error {
 // ARGV[3] -> task message data
 // ARGV[4] -> task timeout in seconds (0 if not timeout)
 // ARGV[5] -> task deadline in unix time (0 if no deadline)
+// ARGV[6] -> current unix time in nsec
 //
 // Output:
 // Returns 1 if successfully enqueued
@@ -149,6 +165,7 @@ redis.call("HSET", KEYS[2],
            "state", "pending",
            "timeout", ARGV[4],
            "deadline", ARGV[5],
+           "pending_since", ARGV[6],
            "unique_key", KEYS[1])
 redis.call("LPUSH", KEYS[3], ARGV[1])
 return 1
@@ -176,6 +193,7 @@ func (r *RDB) EnqueueUnique(ctx context.Context, msg *base.TaskMessage, ttl time
 		encoded,
 		msg.Timeout,
 		msg.Deadline,
+		r.clock.Now().UnixNano(),
 	}
 	n, err := r.runScriptWithErrorCode(ctx, op, enqueueUniqueCmd, keys, argv...)
 	if err != nil {
@@ -214,6 +232,7 @@ if redis.call("EXISTS", KEYS[2]) == 0 then
 	if id then
 		local key = ARGV[2] .. id
 		redis.call("HSET", key, "state", "active")
+		redis.call("HDEL", key, "pending_since")
 		local data = redis.call("HMGET", key, "msg", "timeout", "deadline")
 		local msg = data[1]	
 		local timeout = tonumber(data[2])
@@ -248,7 +267,7 @@ func (r *RDB) Dequeue(qnames ...string) (msg *base.TaskMessage, deadline time.Ti
 			base.DeadlinesKey(qname),
 		}
 		argv := []interface{}{
-			time.Now().Unix(),
+			r.clock.Now().Unix(),
 			base.TaskKeyPrefix(qname),
 		}
 		res, err := dequeueCmd.Run(context.Background(), r.client, keys, argv...).Result()
@@ -335,7 +354,7 @@ return redis.status_reply("OK")
 func (r *RDB) Done(msg *base.TaskMessage) error {
 	var op errors.Op = "rdb.Done"
 	ctx := context.Background()
-	now := time.Now()
+	now := r.clock.Now()
 	expireAt := now.Add(statsTTL)
 	keys := []string{
 		base.ActiveKey(msg.Queue),
@@ -418,7 +437,7 @@ return redis.status_reply("OK")
 func (r *RDB) MarkAsComplete(msg *base.TaskMessage) error {
 	var op errors.Op = "rdb.MarkAsComplete"
 	ctx := context.Background()
-	now := time.Now()
+	now := r.clock.Now()
 	statsExpireAt := now.Add(statsTTL)
 	msg.CompletedAt = now.Unix()
 	encoded, err := base.EncodeMessage(msg)
@@ -638,7 +657,7 @@ return redis.status_reply("OK")`)
 func (r *RDB) Retry(msg *base.TaskMessage, processAt time.Time, errMsg string, isFailure bool) error {
 	var op errors.Op = "rdb.Retry"
 	ctx := context.Background()
-	now := time.Now()
+	now := r.clock.Now()
 	modified := *msg
 	if isFailure {
 		modified.Retried++
@@ -711,7 +730,7 @@ return redis.status_reply("OK")`)
 func (r *RDB) Archive(msg *base.TaskMessage, errMsg string) error {
 	var op errors.Op = "rdb.Archive"
 	ctx := context.Background()
-	now := time.Now()
+	now := r.clock.Now()
 	modified := *msg
 	modified.ErrorMsg = errMsg
 	modified.LastFailedAt = now.Unix()
@@ -754,23 +773,27 @@ func (r *RDB) ForwardIfReady(qnames ...string) error {
 
 // KEYS[1] -> source queue (e.g. asynq:{<qname>:scheduled or asynq:{<qname>}:retry})
 // KEYS[2] -> asynq:{<qname>}:pending
-// ARGV[1] -> current unix time
+// ARGV[1] -> current unix time in seconds
 // ARGV[2] -> task key prefix
+// ARGV[3] -> current unix time in nsec
 // Note: Script moves tasks up to 100 at a time to keep the runtime of script short.
 var forwardCmd = redis.NewScript(`
 local ids = redis.call("ZRANGEBYSCORE", KEYS[1], "-inf", ARGV[1], "LIMIT", 0, 100)
 for _, id in ipairs(ids) do
 	redis.call("LPUSH", KEYS[2], id)
 	redis.call("ZREM", KEYS[1], id)
-	redis.call("HSET", ARGV[2] .. id, "state", "pending")
+	redis.call("HSET", ARGV[2] .. id,
+               "state", "pending",
+               "pending_since", ARGV[3])
 end
 return table.getn(ids)`)
 
 // forward moves tasks with a score less than the current unix time
 // from the src zset to the dst list. It returns the number of tasks moved.
 func (r *RDB) forward(src, dst, taskKeyPrefix string) (int, error) {
-	now := float64(time.Now().Unix())
-	res, err := forwardCmd.Run(context.Background(), r.client, []string{src, dst}, now, taskKeyPrefix).Result()
+	now := r.clock.Now()
+	res, err := forwardCmd.Run(context.Background(), r.client,
+		[]string{src, dst}, now.Unix(), taskKeyPrefix, now.UnixNano()).Result()
 	if err != nil {
 		return 0, errors.E(errors.Internal, fmt.Sprintf("redis eval error: %v", err))
 	}
@@ -835,7 +858,7 @@ func (r *RDB) deleteExpiredCompletedTasks(qname string, batchSize int) (int64, e
 	var op errors.Op = "rdb.DeleteExpiredCompletedTasks"
 	keys := []string{base.CompletedKey(qname)}
 	argv := []interface{}{
-		time.Now().Unix(),
+		r.clock.Now().Unix(),
 		base.TaskKeyPrefix(qname),
 		batchSize,
 	}
@@ -913,7 +936,7 @@ func (r *RDB) WriteServerState(info *base.ServerInfo, workers []*base.WorkerInfo
 	if err != nil {
 		return errors.E(op, errors.Internal, fmt.Sprintf("cannot encode server info: %v", err))
 	}
-	exp := time.Now().Add(ttl).UTC()
+	exp := r.clock.Now().Add(ttl).UTC()
 	args := []interface{}{ttl.Seconds(), bytes} // args to the lua script
 	for _, w := range workers {
 		bytes, err := base.EncodeWorkerInfo(w)
@@ -978,7 +1001,7 @@ func (r *RDB) WriteSchedulerEntries(schedulerID string, entries []*base.Schedule
 		}
 		args = append(args, bytes)
 	}
-	exp := time.Now().Add(ttl).UTC()
+	exp := r.clock.Now().Add(ttl).UTC()
 	key := base.SchedulerEntriesKey(schedulerID)
 	err := r.client.ZAdd(ctx, base.AllSchedulers, &redis.Z{Score: float64(exp.Unix()), Member: key}).Err()
 	if err != nil {
diff --git a/internal/timeutil/timeutil.go b/internal/timeutil/timeutil.go
new file mode 100644
index 0000000..65691ad
--- /dev/null
+++ b/internal/timeutil/timeutil.go
@@ -0,0 +1,38 @@
+// Package timeutil exports functions and types related to time and date.
+package timeutil
+
+import "time"
+
+// A Clock is an object that can tell you the current time.
+//
+// This interface allows decoupling code that uses time from the code that creates
+// a point in time. You can use this to your advantage by injecting Clocks into interfaces
+// rather than having implementations call time.Now() directly.
+//
+// Use RealClock() in production.
+// Use SimulatedClock() in test.
+type Clock interface {
+	Now() time.Time
+}
+
+func NewRealClock() Clock { return &realTimeClock{} }
+
+type realTimeClock struct{}
+
+func (_ *realTimeClock) Now() time.Time { return time.Now() }
+
+// A SimulatedClock is a concrete Clock implementation that doesn't "tick" on its own.
+// Time is advanced by explicit call to the AdvanceTime() or SetTime() functions.
+type SimulatedClock struct {
+	t time.Time
+}
+
+func NewSimulatedClock(t time.Time) *SimulatedClock {
+	return &SimulatedClock{t}
+}
+
+func (c *SimulatedClock) Now() time.Time { return c.t }
+
+func (c *SimulatedClock) SetTime(t time.Time) { c.t = t }
+
+func (c *SimulatedClock) AdvanceTime(d time.Duration) { c.t.Add(d) }


Test patch:
diff --git a/inspector_test.go b/inspector_test.go
index 8a418f1..d3f86fc 100644
--- a/inspector_test.go
+++ b/inspector_test.go
@@ -19,6 +19,7 @@ import (
 	h "github.com/hibiken/asynq/internal/asynqtest"
 	"github.com/hibiken/asynq/internal/base"
 	"github.com/hibiken/asynq/internal/rdb"
+	"github.com/hibiken/asynq/internal/timeutil"
 )
 
 func TestInspectorQueues(t *testing.T) {
@@ -269,18 +270,20 @@ func TestInspectorGetQueueInfo(t *testing.T) {
 	ignoreMemUsg := cmpopts.IgnoreFields(QueueInfo{}, "MemoryUsage")
 
 	inspector := NewInspector(getRedisConnOpt(t))
+	inspector.rdb.SetClock(timeutil.NewSimulatedClock(now))
 
 	tests := []struct {
-		pending   map[string][]*base.TaskMessage
-		active    map[string][]*base.TaskMessage
-		scheduled map[string][]base.Z
-		retry     map[string][]base.Z
-		archived  map[string][]base.Z
-		completed map[string][]base.Z
-		processed map[string]int
-		failed    map[string]int
-		qname     string
-		want      *QueueInfo
+		pending                         map[string][]*base.TaskMessage
+		active                          map[string][]*base.TaskMessage
+		scheduled                       map[string][]base.Z
+		retry                           map[string][]base.Z
+		archived                        map[string][]base.Z
+		completed                       map[string][]base.Z
+		processed                       map[string]int
+		failed                          map[string]int
+		oldestPendingMessageEnqueueTime map[string]time.Time
+		qname                           string
+		want                            *QueueInfo
 	}{
 		{
 			pending: map[string][]*base.TaskMessage{
@@ -326,9 +329,15 @@ func TestInspectorGetQueueInfo(t *testing.T) {
 				"critical": 0,
 				"low":      5,
 			},
+			oldestPendingMessageEnqueueTime: map[string]time.Time{
+				"default":  now.Add(-15 * time.Second),
+				"critical": now.Add(-200 * time.Millisecond),
+				"low":      now.Add(-30 * time.Second),
+			},
 			qname: "default",
 			want: &QueueInfo{
 				Queue:     "default",
+				Latency:   15 * time.Second,
 				Size:      4,
 				Pending:   1,
 				Active:    1,
@@ -352,13 +361,21 @@ func TestInspectorGetQueueInfo(t *testing.T) {
 		h.SeedAllRetryQueues(t, r, tc.retry)
 		h.SeedAllArchivedQueues(t, r, tc.archived)
 		h.SeedAllCompletedQueues(t, r, tc.completed)
+		ctx := context.Background()
 		for qname, n := range tc.processed {
 			processedKey := base.ProcessedKey(qname, now)
-			r.Set(context.Background(), processedKey, n, 0)
+			r.Set(ctx, processedKey, n, 0)
 		}
 		for qname, n := range tc.failed {
 			failedKey := base.FailedKey(qname, now)
-			r.Set(context.Background(), failedKey, n, 0)
+			r.Set(ctx, failedKey, n, 0)
+		}
+		for qname, enqueueTime := range tc.oldestPendingMessageEnqueueTime {
+			if enqueueTime.IsZero() {
+				continue
+			}
+			oldestPendingMessageID := r.LRange(ctx, base.PendingKey(qname), -1, -1).Val()[0] // get the right most msg in the list
+			r.HSet(ctx, base.TaskKey(qname, oldestPendingMessageID), "pending_since", enqueueTime.UnixNano())
 		}
 
 		got, err := inspector.GetQueueInfo(tc.qname)
diff --git a/internal/rdb/inspect_test.go b/internal/rdb/inspect_test.go
index 1a9f49f..8af2584 100644
--- a/internal/rdb/inspect_test.go
+++ b/internal/rdb/inspect_test.go
@@ -17,6 +17,7 @@ import (
 	h "github.com/hibiken/asynq/internal/asynqtest"
 	"github.com/hibiken/asynq/internal/base"
 	"github.com/hibiken/asynq/internal/errors"
+	"github.com/hibiken/asynq/internal/timeutil"
 )
 
 func TestAllQueues(t *testing.T) {
@@ -60,19 +61,21 @@ func TestCurrentStats(t *testing.T) {
 	m5 := h.NewTaskMessageWithQueue("important_notification", nil, "critical")
 	m6 := h.NewTaskMessageWithQueue("minor_notification", nil, "low")
 	now := time.Now()
+	r.SetClock(timeutil.NewSimulatedClock(now))
 
 	tests := []struct {
-		pending    map[string][]*base.TaskMessage
-		inProgress map[string][]*base.TaskMessage
-		scheduled  map[string][]base.Z
-		retry      map[string][]base.Z
-		archived   map[string][]base.Z
-		completed  map[string][]base.Z
-		processed  map[string]int
-		failed     map[string]int
-		paused     []string
-		qname      string
-		want       *Stats
+		pending                         map[string][]*base.TaskMessage
+		active                          map[string][]*base.TaskMessage
+		scheduled                       map[string][]base.Z
+		retry                           map[string][]base.Z
+		archived                        map[string][]base.Z
+		completed                       map[string][]base.Z
+		processed                       map[string]int
+		failed                          map[string]int
+		paused                          []string
+		oldestPendingMessageEnqueueTime map[string]time.Time
+		qname                           string
+		want                            *Stats
 	}{
 		{
 			pending: map[string][]*base.TaskMessage{
@@ -80,7 +83,7 @@ func TestCurrentStats(t *testing.T) {
 				"critical": {m5},
 				"low":      {m6},
 			},
-			inProgress: map[string][]*base.TaskMessage{
+			active: map[string][]*base.TaskMessage{
 				"default":  {m2},
 				"critical": {},
 				"low":      {},
@@ -118,6 +121,11 @@ func TestCurrentStats(t *testing.T) {
 				"critical": 0,
 				"low":      1,
 			},
+			oldestPendingMessageEnqueueTime: map[string]time.Time{
+				"default":  now.Add(-15 * time.Second),
+				"critical": now.Add(-200 * time.Millisecond),
+				"low":      now.Add(-30 * time.Second),
+			},
 			paused: []string{},
 			qname:  "default",
 			want: &Stats{
@@ -132,16 +140,17 @@ func TestCurrentStats(t *testing.T) {
 				Completed: 0,
 				Processed: 120,
 				Failed:    2,
+				Latency:   15 * time.Second,
 				Timestamp: now,
 			},
 		},
 		{
 			pending: map[string][]*base.TaskMessage{
 				"default":  {m1},
-				"critical": {m5},
+				"critical": {},
 				"low":      {m6},
 			},
-			inProgress: map[string][]*base.TaskMessage{
+			active: map[string][]*base.TaskMessage{
 				"default":  {m2},
 				"critical": {},
 				"low":      {},
@@ -179,13 +188,18 @@ func TestCurrentStats(t *testing.T) {
 				"critical": 0,
 				"low":      1,
 			},
+			oldestPendingMessageEnqueueTime: map[string]time.Time{
+				"default":  now.Add(-15 * time.Second),
+				"critical": time.Time{}, // zero value since there's no pending task in this queue
+				"low":      now.Add(-30 * time.Second),
+			},
 			paused: []string{"critical", "low"},
 			qname:  "critical",
 			want: &Stats{
 				Queue:     "critical",
 				Paused:    true,
-				Size:      1,
-				Pending:   1,
+				Size:      0,
+				Pending:   0,
 				Active:    0,
 				Scheduled: 0,
 				Retry:     0,
@@ -193,6 +207,7 @@ func TestCurrentStats(t *testing.T) {
 				Completed: 0,
 				Processed: 100,
 				Failed:    0,
+				Latency:   0,
 				Timestamp: now,
 			},
 		},
@@ -206,18 +221,26 @@ func TestCurrentStats(t *testing.T) {
 			}
 		}
 		h.SeedAllPendingQueues(t, r.client, tc.pending)
-		h.SeedAllActiveQueues(t, r.client, tc.inProgress)
+		h.SeedAllActiveQueues(t, r.client, tc.active)
 		h.SeedAllScheduledQueues(t, r.client, tc.scheduled)
 		h.SeedAllRetryQueues(t, r.client, tc.retry)
 		h.SeedAllArchivedQueues(t, r.client, tc.archived)
 		h.SeedAllCompletedQueues(t, r.client, tc.completed)
+		ctx := context.Background()
 		for qname, n := range tc.processed {
 			processedKey := base.ProcessedKey(qname, now)
-			r.client.Set(context.Background(), processedKey, n, 0)
+			r.client.Set(ctx, processedKey, n, 0)
 		}
 		for qname, n := range tc.failed {
 			failedKey := base.FailedKey(qname, now)
-			r.client.Set(context.Background(), failedKey, n, 0)
+			r.client.Set(ctx, failedKey, n, 0)
+		}
+		for qname, enqueueTime := range tc.oldestPendingMessageEnqueueTime {
+			if enqueueTime.IsZero() {
+				continue
+			}
+			oldestPendingMessageID := r.client.LRange(ctx, base.PendingKey(qname), -1, -1).Val()[0] // get the right most msg in the list
+			r.client.HSet(ctx, base.TaskKey(qname, oldestPendingMessageID), "pending_since", enqueueTime.UnixNano())
 		}
 
 		got, err := r.CurrentStats(tc.qname)
diff --git a/internal/rdb/rdb_test.go b/internal/rdb/rdb_test.go
index 8d6a464..caacab5 100644
--- a/internal/rdb/rdb_test.go
+++ b/internal/rdb/rdb_test.go
@@ -21,6 +21,7 @@ import (
 	h "github.com/hibiken/asynq/internal/asynqtest"
 	"github.com/hibiken/asynq/internal/base"
 	"github.com/hibiken/asynq/internal/errors"
+	"github.com/hibiken/asynq/internal/timeutil"
 )
 
 // variables used for package testing.
@@ -67,6 +68,9 @@ func TestEnqueue(t *testing.T) {
 	t2 := h.NewTaskMessageWithQueue("generate_csv", h.JSON(map[string]interface{}{}), "csv")
 	t3 := h.NewTaskMessageWithQueue("sync", nil, "low")
 
+	enqueueTime := time.Now()
+	r.SetClock(timeutil.NewSimulatedClock(enqueueTime))
+
 	tests := []struct {
 		msg *base.TaskMessage
 	}{
@@ -115,6 +119,10 @@ func TestEnqueue(t *testing.T) {
 		if want := strconv.Itoa(int(tc.msg.Deadline)); deadline != want {
 			t.Errorf("deadline field under task-key is set to %v, want %v", deadline, want)
 		}
+		pendingSince := r.client.HGet(context.Background(), taskKey, "pending_since").Val() // "pending_since" field
+		if want := strconv.Itoa(int(enqueueTime.UnixNano())); pendingSince != want {
+			t.Errorf("pending_since field under task-key is set to %v, want %v", pendingSince, want)
+		}
 
 		// Check queue is in the AllQueues set.
 		if !r.client.SIsMember(context.Background(), base.AllQueues, tc.msg.Queue).Val() {
@@ -170,6 +178,9 @@ func TestEnqueueUnique(t *testing.T) {
 		UniqueKey: base.UniqueKey(base.DefaultQueueName, "email", h.JSON(map[string]interface{}{"user_id": 123})),
 	}
 
+	enqueueTime := time.Now()
+	r.SetClock(timeutil.NewSimulatedClock(enqueueTime))
+
 	tests := []struct {
 		msg *base.TaskMessage
 		ttl time.Duration // uniqueness ttl
@@ -230,6 +241,10 @@ func TestEnqueueUnique(t *testing.T) {
 		if want := strconv.Itoa(int(tc.msg.Deadline)); deadline != want {
 			t.Errorf("deadline field under task-key is set to %v, want %v", deadline, want)
 		}
+		pendingSince := r.client.HGet(context.Background(), taskKey, "pending_since").Val() // "pending_since" field
+		if want := strconv.Itoa(int(enqueueTime.UnixNano())); pendingSince != want {
+			t.Errorf("pending_since field under task-key is set to %v, want %v", pendingSince, want)
+		}
 		uniqueKey := r.client.HGet(context.Background(), taskKey, "unique_key").Val() // "unique_key" field
 		if uniqueKey != tc.msg.UniqueKey {
 			t.Errorf("uniqueue_key field under task key is set to %q, want %q", uniqueKey, tc.msg.UniqueKey)
@@ -2055,6 +2070,9 @@ func TestForwardIfReady(t *testing.T) {
 		h.SeedAllScheduledQueues(t, r.client, tc.scheduled)
 		h.SeedAllRetryQueues(t, r.client, tc.retry)
 
+		now := time.Now()
+		r.SetClock(timeutil.NewSimulatedClock(now))
+
 		err := r.ForwardIfReady(tc.qnames...)
 		if err != nil {
 			t.Errorf("(*RDB).CheckScheduled(%v) = %v, want nil", tc.qnames, err)
@@ -2066,6 +2084,13 @@ func TestForwardIfReady(t *testing.T) {
 			if diff := cmp.Diff(want, gotPending, h.SortMsgOpt); diff != "" {
 				t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.PendingKey(qname), diff)
 			}
+			// Make sure "pending_since" field is set
+			for _, msg := range gotPending {
+				pendingSince := r.client.HGet(context.Background(), base.TaskKey(msg.Queue, msg.ID), "pending_since").Val()
+				if want := strconv.Itoa(int(now.UnixNano())); pendingSince != want {
+					t.Error("pending_since field is not set for newly pending message")
+				}
+			}
 		}
 		for qname, want := range tc.wantScheduled {
 			gotScheduled := h.GetScheduledMessages(t, r.client, qname)