Repository: hibiken/asynq
Pull request #174 (2020-06-22 15:46:28)

Automatic recovery of orphan tasks
Currently when server crashes while processing tasks, the tasks get stuck in in-progress queue forever.

This PR introduces recovery mechanism for such cases.

Changes:

- All tasks will have deadline (specified via Timeout or Deadline option, by default if none are specified it'll use 30 mins timeout)
- Tasks which exceeded its deadline will be retried, and treated as failed.

TODO:

- [x] Document default 30mins timeout
- [x] Changelog
- [x] Expose `migrate` command to migrate all tasks enqueued with previous version to smooth the version upgrade

codecov[bot]: # [Codecov](https://codecov.io/gh/hibiken/asynq/pull/174?src=pr&el=h1) Report
> Merging [#174](https://codecov.io/gh/hibiken/asynq/pull/174?src=pr&el=desc) into [next](https://codecov.io/gh/hibiken/asynq/commit/4e5f5969108de59ffd2574471cd2e3399a2f4914&el=desc) will **decrease** coverage by `1.01%`.
> The diff coverage is `77.33%`.

[![Impacted file tree graph](https://codecov.io/gh/hibiken/asynq/pull/174/graphs/tree.svg?width=650&height=150&src=pr&token=BAKLSWL2G3)](https://codecov.io/gh/hibiken/asynq/pull/174?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##             next     #174      +/-   ##
==========================================
- Coverage   84.04%   83.02%   -1.02%     
==========================================
  Files          16       17       +1     
  Lines        1598     1685      +87     
==========================================
+ Hits         1343     1399      +56     
- Misses        172      193      +21     
- Partials       83       93      +10     
```


| [Impacted Files](https://codecov.io/gh/hibiken/asynq/pull/174?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [internal/base/base.go](https://codecov.io/gh/hibiken/asynq/pull/174/diff?src=pr&el=tree#diff-aW50ZXJuYWwvYmFzZS9iYXNlLmdv) | `90.90% <ø> (-1.03%)` | :arrow_down: |
| [internal/rdb/rdb.go](https://codecov.io/gh/hibiken/asynq/pull/174/diff?src=pr&el=tree#diff-aW50ZXJuYWwvcmRiL3JkYi5nbw==) | `70.83% <64.28%> (-3.56%)` | :arrow_down: |
| [processor.go](https://codecov.io/gh/hibiken/asynq/pull/174/diff?src=pr&el=tree#diff-cHJvY2Vzc29yLmdv) | `85.22% <72.09%> (-5.26%)` | :arrow_down: |
| [recoverer.go](https://codecov.io/gh/hibiken/asynq/pull/174/diff?src=pr&el=tree#diff-cmVjb3ZlcmVyLmdv) | `82.92% <82.92%> (ø)` | |
| [client.go](https://codecov.io/gh/hibiken/asynq/pull/174/diff?src=pr&el=tree#diff-Y2xpZW50Lmdv) | `97.27% <100.00%> (+0.21%)` | :arrow_up: |
| [context.go](https://codecov.io/gh/hibiken/asynq/pull/174/diff?src=pr&el=tree#diff-Y29udGV4dC5nbw==) | `100.00% <100.00%> (ø)` | |
| [server.go](https://codecov.io/gh/hibiken/asynq/pull/174/diff?src=pr&el=tree#diff-c2VydmVyLmdv) | `85.22% <100.00%> (+0.79%)` | :arrow_up: |
| [syncer.go](https://codecov.io/gh/hibiken/asynq/pull/174/diff?src=pr&el=tree#diff-c3luY2VyLmdv) | `100.00% <100.00%> (+6.66%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/hibiken/asynq/pull/174?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/hibiken/asynq/pull/174?src=pr&el=footer). Last update [4e5f596...fb88618](https://codecov.io/gh/hibiken/asynq/pull/174?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 33aecb8..c3942fb 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed
+
+- All tasks now requires timeout or deadline. By default, timeout is set to 30 mins.
+- Tasks that exceed its deadline are automatically retried.
+- Encoding schema for task message has changed. Please install the lastest CLI and run `migrate` command if 
+  you have tasks enqueued by the previous version of asynq.
+
 ## [0.9.4] - 2020-06-13
 
 ### Fixed
diff --git a/client.go b/client.go
index af0523a..2492633 100644
--- a/client.go
+++ b/client.go
@@ -69,13 +69,23 @@ func Queue(name string) Option {
 }
 
 // Timeout returns an option to specify how long a task may run.
+// If the timeout elapses before the Handler returns, then the task
+// will be retried.
 //
 // Zero duration means no limit.
+//
+// If there's a conflicting Deadline option, whichever comes earliest
+// will be used.
 func Timeout(d time.Duration) Option {
 	return timeoutOption(d)
 }
 
 // Deadline returns an option to specify the deadline for the given task.
+// If it reaches the deadline before the Handler returns, then the task
+// will be retried.
+//
+// If there's a conflicting Timeout option, whichever comes earliest
+// will be used.
 func Deadline(t time.Time) Option {
 	return deadlineOption(t)
 }
@@ -110,7 +120,7 @@ func composeOptions(opts ...Option) option {
 	res := option{
 		retry:    defaultMaxRetry,
 		queue:    base.DefaultQueueName,
-		timeout:  0,
+		timeout:  0, // do not set to deafultTimeout here
 		deadline: time.Time{},
 	}
 	for _, opt := range opts {
@@ -165,8 +175,19 @@ func serializePayload(payload map[string]interface{}) string {
 	return b.String()
 }
 
-// Default max retry count used if nothing is specified.
-const defaultMaxRetry = 25
+const (
+	// Default max retry count used if nothing is specified.
+	defaultMaxRetry = 25
+
+	// Default timeout used if both timeout and deadline are not specified.
+	defaultTimeout = 30 * time.Minute
+)
+
+// Value zero indicates no timeout and no deadline.
+var (
+	noTimeout  time.Duration = 0
+	noDeadline time.Time     = time.Unix(0, 0)
+)
 
 // SetDefaultOptions sets options to be used for a given task type.
 // The argument opts specifies the behavior of task processing.
@@ -185,6 +206,7 @@ func (c *Client) SetDefaultOptions(taskType string, opts ...Option) {
 //
 // The argument opts specifies the behavior of task processing.
 // If there are conflicting Option values the last one overrides others.
+// By deafult, max retry is set to 25 and timeout is set to 30 minutes.
 func (c *Client) EnqueueAt(t time.Time, task *Task, opts ...Option) error {
 	return c.enqueueAt(t, task, opts...)
 }
@@ -195,6 +217,7 @@ func (c *Client) EnqueueAt(t time.Time, task *Task, opts ...Option) error {
 //
 // The argument opts specifies the behavior of task processing.
 // If there are conflicting Option values the last one overrides others.
+// By deafult, max retry is set to 25 and timeout is set to 30 minutes.
 func (c *Client) Enqueue(task *Task, opts ...Option) error {
 	return c.enqueueAt(time.Now(), task, opts...)
 }
@@ -205,6 +228,7 @@ func (c *Client) Enqueue(task *Task, opts ...Option) error {
 //
 // The argument opts specifies the behavior of task processing.
 // If there are conflicting Option values the last one overrides others.
+// By deafult, max retry is set to 25 and timeout is set to 30 minutes.
 func (c *Client) EnqueueIn(d time.Duration, task *Task, opts ...Option) error {
 	return c.enqueueAt(time.Now().Add(d), task, opts...)
 }
@@ -221,14 +245,26 @@ func (c *Client) enqueueAt(t time.Time, task *Task, opts ...Option) error {
 		opts = append(defaults, opts...)
 	}
 	opt := composeOptions(opts...)
+	deadline := noDeadline
+	if !opt.deadline.IsZero() {
+		deadline = opt.deadline
+	}
+	timeout := noTimeout
+	if opt.timeout != 0 {
+		timeout = opt.timeout
+	}
+	if deadline.Equal(noDeadline) && timeout == noTimeout {
+		// If neither deadline nor timeout are set, use default timeout.
+		timeout = defaultTimeout
+	}
 	msg := &base.TaskMessage{
 		ID:        xid.New(),
 		Type:      task.Type,
 		Payload:   task.Payload.data,
 		Queue:     opt.queue,
 		Retry:     opt.retry,
-		Timeout:   opt.timeout.String(),
-		Deadline:  opt.deadline.Format(time.RFC3339),
+		Deadline:  deadline.Unix(),
+		Timeout:   int64(timeout.Seconds()),
 		UniqueKey: uniqueKey(task, opt.uniqueTTL, opt.queue),
 	}
 	var err error
diff --git a/context.go b/context.go
index ee69ba1..1ed6634 100644
--- a/context.go
+++ b/context.go
@@ -27,25 +27,14 @@ type ctxKey int
 const metadataCtxKey ctxKey = 0
 
 // createContext returns a context and cancel function for a given task message.
-func createContext(msg *base.TaskMessage) (ctx context.Context, cancel context.CancelFunc) {
+func createContext(msg *base.TaskMessage, deadline time.Time) (context.Context, context.CancelFunc) {
 	metadata := taskMetadata{
 		id:         msg.ID.String(),
 		maxRetry:   msg.Retry,
 		retryCount: msg.Retried,
 	}
-	ctx = context.WithValue(context.Background(), metadataCtxKey, metadata)
-	timeout, err := time.ParseDuration(msg.Timeout)
-	if err == nil && timeout != 0 {
-		ctx, cancel = context.WithTimeout(ctx, timeout)
-	}
-	deadline, err := time.Parse(time.RFC3339, msg.Deadline)
-	if err == nil && !deadline.IsZero() {
-		ctx, cancel = context.WithDeadline(ctx, deadline)
-	}
-	if cancel == nil {
-		ctx, cancel = context.WithCancel(ctx)
-	}
-	return ctx, cancel
+	ctx := context.WithValue(context.Background(), metadataCtxKey, metadata)
+	return context.WithDeadline(ctx, deadline)
 }
 
 // GetTaskID extracts a task ID from a context, if any.
diff --git a/internal/base/base.go b/internal/base/base.go
index 4733aa8..f559589 100644
--- a/internal/base/base.go
+++ b/internal/base/base.go
@@ -35,6 +35,7 @@ const (
 	RetryQueue      = "asynq:retry"                  // ZSET
 	DeadQueue       = "asynq:dead"                   // ZSET
 	InProgressQueue = "asynq:in_progress"            // LIST
+	KeyDeadlines    = "asynq:deadlines"              // ZSET
 	PausedQueues    = "asynq:paused"                 // SET
 	CancelChannel   = "asynq:cancel"                 // PubSub channel
 )
@@ -88,18 +89,20 @@ type TaskMessage struct {
 	// ErrorMsg holds the error message from the last failure.
 	ErrorMsg string
 
-	// Timeout specifies how long a task may run.
-	// The string value should be compatible with time.Duration.ParseDuration.
+	// Timeout specifies timeout in seconds.
+	// If task processing doesn't complete within the timeout, the task will be retried
+	// if retry count is remaining. Otherwise it will be moved to the dead queue.
 	//
-	// Zero means no limit.
-	Timeout string
+	// Use zero to indicate no timeout.
+	Timeout int64
 
-	// Deadline specifies the deadline for the task.
-	// Task won't be processed if it exceeded its deadline.
-	// The string shoulbe be in RFC3339 format.
+	// Deadline specifies the deadline for the task in Unix time,
+	// the number of seconds elapsed since January 1, 1970 UTC.
+	// If task processing doesn't complete before the deadline, the task will be retried
+	// if retry count is remaining. Otherwise it will be moved to the dead queue.
 	//
-	// time.Time's zero value means no deadline.
-	Deadline string
+	// Use zero to indicate no deadline.
+	Deadline int64
 
 	// UniqueKey holds the redis key used for uniqueness lock for this task.
 	//
@@ -247,24 +250,13 @@ func (c *Cancelations) Get(id string) (fn context.CancelFunc, ok bool) {
 	return fn, ok
 }
 
-// GetAll returns all cancel funcs.
-func (c *Cancelations) GetAll() []context.CancelFunc {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-	var res []context.CancelFunc
-	for _, fn := range c.cancelFuncs {
-		res = append(res, fn)
-	}
-	return res
-}
-
 // Broker is a message broker that supports operations to manage task queues.
 //
 // See rdb.RDB as a reference implementation.
 type Broker interface {
 	Enqueue(msg *TaskMessage) error
 	EnqueueUnique(msg *TaskMessage, ttl time.Duration) error
-	Dequeue(qnames ...string) (*TaskMessage, error)
+	Dequeue(qnames ...string) (*TaskMessage, time.Time, error)
 	Done(msg *TaskMessage) error
 	Requeue(msg *TaskMessage) error
 	Schedule(msg *TaskMessage, processAt time.Time) error
@@ -272,6 +264,7 @@ type Broker interface {
 	Retry(msg *TaskMessage, processAt time.Time, errMsg string) error
 	Kill(msg *TaskMessage, errMsg string) error
 	CheckAndEnqueue() error
+	ListDeadlineExceeded(deadline time.Time) ([]*TaskMessage, error)
 	WriteServerState(info *ServerInfo, workers []*WorkerInfo, ttl time.Duration) error
 	ClearServerState(host string, pid int, serverID string) error
 	CancelationPubSub() (*redis.PubSub, error) // TODO: Need to decouple from redis to support other brokers
diff --git a/internal/rdb/rdb.go b/internal/rdb/rdb.go
index af71be9..e33f11e 100644
--- a/internal/rdb/rdb.go
+++ b/internal/rdb/rdb.go
@@ -9,6 +9,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"strconv"
 	"time"
 
 	"github.com/go-redis/redis/v7"
@@ -102,68 +103,110 @@ func (r *RDB) EnqueueUnique(msg *base.TaskMessage, ttl time.Duration) error {
 	return nil
 }
 
-// Dequeue queries given queues in order and pops a task message if there is one and returns it.
+// Dequeue queries given queues in order and pops a task message
+// off a queue if one exists and returns the message and deadline.
 // Dequeue skips a queue if the queue is paused.
 // If all queues are empty, ErrNoProcessableTask error is returned.
-func (r *RDB) Dequeue(qnames ...string) (*base.TaskMessage, error) {
+func (r *RDB) Dequeue(qnames ...string) (msg *base.TaskMessage, deadline time.Time, err error) {
 	var qkeys []interface{}
 	for _, q := range qnames {
 		qkeys = append(qkeys, base.QueueKey(q))
 	}
-	data, err := r.dequeue(qkeys...)
+	data, d, err := r.dequeue(qkeys...)
 	if err == redis.Nil {
-		return nil, ErrNoProcessableTask
+		return nil, time.Time{}, ErrNoProcessableTask
 	}
 	if err != nil {
-		return nil, err
+		return nil, time.Time{}, err
+	}
+	if msg, err = base.DecodeMessage(data); err != nil {
+		return nil, time.Time{}, err
 	}
-	return base.DecodeMessage(data)
+	return msg, time.Unix(d, 0), nil
 }
 
-// KEYS[1] -> asynq:in_progress
-// KEYS[2] -> asynq:paused
-// ARGV    -> List of queues to query in order
+// KEYS[1]  -> asynq:in_progress
+// KEYS[2]  -> asynq:paused
+// KEYS[3]  -> asynq:deadlines
+// ARGV[1]  -> current time in Unix time
+// ARGV[2:] -> List of queues to query in order
 //
 // dequeueCmd checks whether a queue is paused first, before
 // calling RPOPLPUSH to pop a task from the queue.
+// It computes the task deadline by inspecting Timout and Deadline fields,
+// and inserts the task with deadlines set.
 var dequeueCmd = redis.NewScript(`
-for _, qkey in ipairs(ARGV) do
+for i = 2, table.getn(ARGV) do
+	local qkey = ARGV[i]
 	if redis.call("SISMEMBER", KEYS[2], qkey) == 0 then
-		local res = redis.call("RPOPLPUSH", qkey, KEYS[1])
-		if res then
-			return res
+		local msg = redis.call("RPOPLPUSH", qkey, KEYS[1])
+		if msg then
+			local decoded = cjson.decode(msg)
+			local timeout = decoded["Timeout"]
+			local deadline = decoded["Deadline"]
+			local score
+			if timeout ~= 0 and deadline ~= 0 then
+				score = math.min(ARGV[1]+timeout, deadline)
+			elseif timeout ~= 0 then
+				score = ARGV[1] + timeout
+			elseif deadline ~= 0 then
+				score = deadline
+			else
+				return redis.error_reply("asynq internal error: both timeout and deadline are not set")
+			end
+			redis.call("ZADD", KEYS[3], score, msg)
+			return {msg, score}
 		end
 	end
 end
 return nil`)
 
-func (r *RDB) dequeue(qkeys ...interface{}) (data string, err error) {
+func (r *RDB) dequeue(qkeys ...interface{}) (msgjson string, deadline int64, err error) {
+	var args []interface{}
+	args = append(args, time.Now().Unix())
+	args = append(args, qkeys...)
 	res, err := dequeueCmd.Run(r.client,
-		[]string{base.InProgressQueue, base.PausedQueues}, qkeys...).Result()
+		[]string{base.InProgressQueue, base.PausedQueues, base.KeyDeadlines}, args...).Result()
+	if err != nil {
+		return "", 0, err
+	}
+	data, err := cast.ToSliceE(res)
 	if err != nil {
-		return "", err
+		return "", 0, err
 	}
-	return cast.ToStringE(res)
+	if len(data) != 2 {
+		return "", 0, fmt.Errorf("asynq: internal error: dequeue command returned %d values", len(data))
+	}
+	if msgjson, err = cast.ToStringE(data[0]); err != nil {
+		return "", 0, err
+	}
+	if deadline, err = cast.ToInt64E(data[1]); err != nil {
+		return "", 0, err
+	}
+	return msgjson, deadline, nil
 }
 
 // KEYS[1] -> asynq:in_progress
-// KEYS[2] -> asynq:processed:<yyyy-mm-dd>
-// KEYS[3] -> unique key in the format <type>:<payload>:<qname>
+// KEYS[2] -> asynq:deadlines
+// KEYS[3] -> asynq:processed:<yyyy-mm-dd>
+// KEYS[4] -> unique key in the format <type>:<payload>:<qname>
 // ARGV[1] -> base.TaskMessage value
 // ARGV[2] -> stats expiration timestamp
 // ARGV[3] -> task ID
 // Note: LREM count ZERO means "remove all elements equal to val"
 var doneCmd = redis.NewScript(`
-local x = redis.call("LREM", KEYS[1], 0, ARGV[1]) 
-if x == 0 then
+if redis.call("LREM", KEYS[1], 0, ARGV[1]) == 0 then
   return redis.error_reply("NOT FOUND")
 end
-local n = redis.call("INCR", KEYS[2])
+if redis.call("ZREM", KEYS[2], ARGV[1]) == 0 then
+  return redis.error_reply("NOT FOUND")
+end
+local n = redis.call("INCR", KEYS[3])
 if tonumber(n) == 1 then
-	redis.call("EXPIREAT", KEYS[2], ARGV[2])
+	redis.call("EXPIREAT", KEYS[3], ARGV[2])
 end
-if string.len(KEYS[3]) > 0 and redis.call("GET", KEYS[3]) == ARGV[3] then
-  redis.call("DEL", KEYS[3])
+if string.len(KEYS[4]) > 0 and redis.call("GET", KEYS[4]) == ARGV[3] then
+  redis.call("DEL", KEYS[4])
 end
 return redis.status_reply("OK")
 `)
@@ -179,17 +222,23 @@ func (r *RDB) Done(msg *base.TaskMessage) error {
 	processedKey := base.ProcessedKey(now)
 	expireAt := now.Add(statsTTL)
 	return doneCmd.Run(r.client,
-		[]string{base.InProgressQueue, processedKey, msg.UniqueKey},
+		[]string{base.InProgressQueue, base.KeyDeadlines, processedKey, msg.UniqueKey},
 		encoded, expireAt.Unix(), msg.ID.String()).Err()
 }
 
 // KEYS[1] -> asynq:in_progress
-// KEYS[2] -> asynq:queues:<qname>
+// KEYS[2] -> asynq:deadlines
+// KEYS[3] -> asynq:queues:<qname>
 // ARGV[1] -> base.TaskMessage value
 // Note: Use RPUSH to push to the head of the queue.
 var requeueCmd = redis.NewScript(`
-redis.call("LREM", KEYS[1], 0, ARGV[1])
-redis.call("RPUSH", KEYS[2], ARGV[1])
+if redis.call("LREM", KEYS[1], 0, ARGV[1]) == 0 then
+  return redis.error_reply("NOT FOUND")
+end
+if redis.call("ZREM", KEYS[2], ARGV[1]) == 0 then
+  return redis.error_reply("NOT FOUND")
+end
+redis.call("RPUSH", KEYS[3], ARGV[1])
 return redis.status_reply("OK")`)
 
 // Requeue moves the task from in-progress queue to the specified queue.
@@ -199,7 +248,7 @@ func (r *RDB) Requeue(msg *base.TaskMessage) error {
 		return err
 	}
 	return requeueCmd.Run(r.client,
-		[]string{base.InProgressQueue, base.QueueKey(msg.Queue)},
+		[]string{base.InProgressQueue, base.KeyDeadlines, base.QueueKey(msg.Queue)},
 		encoded).Err()
 }
 
@@ -271,26 +320,29 @@ func (r *RDB) ScheduleUnique(msg *base.TaskMessage, processAt time.Time, ttl tim
 }
 
 // KEYS[1] -> asynq:in_progress
-// KEYS[2] -> asynq:retry
-// KEYS[3] -> asynq:processed:<yyyy-mm-dd>
-// KEYS[4] -> asynq:failure:<yyyy-mm-dd>
+// KEYS[2] -> asynq:deadlines
+// KEYS[3] -> asynq:retry
+// KEYS[4] -> asynq:processed:<yyyy-mm-dd>
+// KEYS[5] -> asynq:failure:<yyyy-mm-dd>
 // ARGV[1] -> base.TaskMessage value to remove from base.InProgressQueue queue
 // ARGV[2] -> base.TaskMessage value to add to Retry queue
 // ARGV[3] -> retry_at UNIX timestamp
 // ARGV[4] -> stats expiration timestamp
 var retryCmd = redis.NewScript(`
-local x = redis.call("LREM", KEYS[1], 0, ARGV[1])
-if x == 0 then
+if redis.call("LREM", KEYS[1], 0, ARGV[1]) == 0 then
   return redis.error_reply("NOT FOUND")
 end
-redis.call("ZADD", KEYS[2], ARGV[3], ARGV[2])
-local n = redis.call("INCR", KEYS[3])
+if redis.call("ZREM", KEYS[2], ARGV[1]) == 0 then
+  return redis.error_reply("NOT FOUND")
+end
+redis.call("ZADD", KEYS[3], ARGV[3], ARGV[2])
+local n = redis.call("INCR", KEYS[4])
 if tonumber(n) == 1 then
-	redis.call("EXPIREAT", KEYS[3], ARGV[4])
+	redis.call("EXPIREAT", KEYS[4], ARGV[4])
 end
-local m = redis.call("INCR", KEYS[4])
+local m = redis.call("INCR", KEYS[5])
 if tonumber(m) == 1 then
-	redis.call("EXPIREAT", KEYS[4], ARGV[4])
+	redis.call("EXPIREAT", KEYS[5], ARGV[4])
 end
 return redis.status_reply("OK")`)
 
@@ -313,7 +365,7 @@ func (r *RDB) Retry(msg *base.TaskMessage, processAt time.Time, errMsg string) e
 	failureKey := base.FailureKey(now)
 	expireAt := now.Add(statsTTL)
 	return retryCmd.Run(r.client,
-		[]string{base.InProgressQueue, base.RetryQueue, processedKey, failureKey},
+		[]string{base.InProgressQueue, base.KeyDeadlines, base.RetryQueue, processedKey, failureKey},
 		msgToRemove, msgToAdd, processAt.Unix(), expireAt.Unix()).Err()
 }
 
@@ -323,9 +375,10 @@ const (
 )
 
 // KEYS[1] -> asynq:in_progress
-// KEYS[2] -> asynq:dead
-// KEYS[3] -> asynq:processed:<yyyy-mm-dd>
-// KEYS[4] -> asynq.failure:<yyyy-mm-dd>
+// KEYS[2] -> asynq:deadlines
+// KEYS[3] -> asynq:dead
+// KEYS[4] -> asynq:processed:<yyyy-mm-dd>
+// KEYS[5] -> asynq.failure:<yyyy-mm-dd>
 // ARGV[1] -> base.TaskMessage value to remove from base.InProgressQueue queue
 // ARGV[2] -> base.TaskMessage value to add to Dead queue
 // ARGV[3] -> died_at UNIX timestamp
@@ -333,20 +386,22 @@ const (
 // ARGV[5] -> max number of tasks in dead queue (e.g., 100)
 // ARGV[6] -> stats expiration timestamp
 var killCmd = redis.NewScript(`
-local x = redis.call("LREM", KEYS[1], 0, ARGV[1])
-if x == 0 then
+if redis.call("LREM", KEYS[1], 0, ARGV[1]) == 0 then
   return redis.error_reply("NOT FOUND")
 end
-redis.call("ZADD", KEYS[2], ARGV[3], ARGV[2])
-redis.call("ZREMRANGEBYSCORE", KEYS[2], "-inf", ARGV[4])
-redis.call("ZREMRANGEBYRANK", KEYS[2], 0, -ARGV[5])
-local n = redis.call("INCR", KEYS[3])
+if redis.call("ZREM", KEYS[2], ARGV[1]) == 0 then
+  return redis.error_reply("NOT FOUND")
+end
+redis.call("ZADD", KEYS[3], ARGV[3], ARGV[2])
+redis.call("ZREMRANGEBYSCORE", KEYS[3], "-inf", ARGV[4])
+redis.call("ZREMRANGEBYRANK", KEYS[3], 0, -ARGV[5])
+local n = redis.call("INCR", KEYS[4])
 if tonumber(n) == 1 then
-	redis.call("EXPIREAT", KEYS[3], ARGV[6])
+	redis.call("EXPIREAT", KEYS[4], ARGV[6])
 end
-local m = redis.call("INCR", KEYS[4])
+local m = redis.call("INCR", KEYS[5])
 if tonumber(m) == 1 then
-	redis.call("EXPIREAT", KEYS[4], ARGV[6])
+	redis.call("EXPIREAT", KEYS[5], ARGV[6])
 end
 return redis.status_reply("OK")`)
 
@@ -370,7 +425,7 @@ func (r *RDB) Kill(msg *base.TaskMessage, errMsg string) error {
 	failureKey := base.FailureKey(now)
 	expireAt := now.Add(statsTTL)
 	return killCmd.Run(r.client,
-		[]string{base.InProgressQueue, base.DeadQueue, processedKey, failureKey},
+		[]string{base.InProgressQueue, base.KeyDeadlines, base.DeadQueue, processedKey, failureKey},
 		msgToRemove, msgToAdd, now.Unix(), limit, maxDeadTasks, expireAt.Unix()).Err()
 }
 
@@ -416,6 +471,27 @@ func (r *RDB) forward(src string) (int, error) {
 	return cast.ToInt(res), nil
 }
 
+// ListDeadlineExceeded returns a list of task messages that have exceeded the given deadline.
+func (r *RDB) ListDeadlineExceeded(deadline time.Time) ([]*base.TaskMessage, error) {
+	var msgs []*base.TaskMessage
+	opt := &redis.ZRangeBy{
+		Min: "-inf",
+		Max: strconv.FormatInt(deadline.Unix(), 10),
+	}
+	res, err := r.client.ZRangeByScore(base.KeyDeadlines, opt).Result()
+	if err != nil {
+		return nil, err
+	}
+	for _, s := range res {
+		msg, err := base.DecodeMessage(s)
+		if err != nil {
+			return nil, err
+		}
+		msgs = append(msgs, msg)
+	}
+	return msgs, nil
+}
+
 // KEYS[1]  -> asynq:servers:<host:pid:sid>
 // KEYS[2]  -> asynq:servers
 // KEYS[3]  -> asynq:workers<host:pid:sid>
@@ -463,7 +539,7 @@ func (r *RDB) WriteServerState(info *base.ServerInfo, workers []*base.WorkerInfo
 // KEYS[2] -> asynq:servers:<host:pid:sid>
 // KEYS[3] -> asynq:workers
 // KEYS[4] -> asynq:workers<host:pid:sid>
-var clearProcessInfoCmd = redis.NewScript(`
+var clearServerStateCmd = redis.NewScript(`
 redis.call("ZREM", KEYS[1], KEYS[2])
 redis.call("DEL", KEYS[2])
 redis.call("ZREM", KEYS[3], KEYS[4])
@@ -474,7 +550,7 @@ return redis.status_reply("OK")`)
 func (r *RDB) ClearServerState(host string, pid int, serverID string) error {
 	skey := base.ServerInfoKey(host, pid, serverID)
 	wkey := base.WorkersKey(host, pid, serverID)
-	return clearProcessInfoCmd.Run(r.client,
+	return clearServerStateCmd.Run(r.client,
 		[]string{base.AllServers, skey, base.AllWorkers, wkey}).Err()
 }
 
diff --git a/processor.go b/processor.go
index 5216a62..459b64d 100644
--- a/processor.go
+++ b/processor.go
@@ -50,12 +50,12 @@ type processor struct {
 	done chan struct{}
 	once sync.Once
 
-	// abort channel is closed when the shutdown of the "processor" goroutine starts.
-	abort chan struct{}
-
-	// quit channel communicates to the in-flight worker goroutines to stop.
+	// quit channel is closed when the shutdown of the "processor" goroutine starts.
 	quit chan struct{}
 
+	// abort channel communicates to the in-flight worker goroutines to stop.
+	abort chan struct{}
+
 	// cancelations is a set of cancel functions for all in-progress tasks.
 	cancelations *base.Cancelations
 
@@ -98,8 +98,8 @@ func newProcessor(params processorParams) *processor {
 		errLogLimiter:  rate.NewLimiter(rate.Every(3*time.Second), 1),
 		sema:           make(chan struct{}, params.concurrency),
 		done:           make(chan struct{}),
-		abort:          make(chan struct{}),
 		quit:           make(chan struct{}),
+		abort:          make(chan struct{}),
 		errHandler:     params.errHandler,
 		handler:        HandlerFunc(func(ctx context.Context, t *Task) error { return fmt.Errorf("handler not set") }),
 		starting:       params.starting,
@@ -113,7 +113,7 @@ func (p *processor) stop() {
 	p.once.Do(func() {
 		p.logger.Debug("Processor shutting down...")
 		// Unblock if processor is waiting for sema token.
-		close(p.abort)
+		close(p.quit)
 		// Signal the processor goroutine to stop processing tasks
 		// from the queue.
 		p.done <- struct{}{}
@@ -124,14 +124,9 @@ func (p *processor) stop() {
 func (p *processor) terminate() {
 	p.stop()
 
-	time.AfterFunc(p.shutdownTimeout, func() { close(p.quit) })
-	p.logger.Info("Waiting for all workers to finish...")
-
-	// send cancellation signal to all in-progress task handlers
-	for _, cancel := range p.cancelations.GetAll() {
-		cancel()
-	}
+	time.AfterFunc(p.shutdownTimeout, func() { close(p.abort) })
 
+	p.logger.Info("Waiting for all workers to finish...")
 	// block until all workers have released the token
 	for i := 0; i < cap(p.sema); i++ {
 		p.sema <- struct{}{}
@@ -158,30 +153,30 @@ func (p *processor) start(wg *sync.WaitGroup) {
 // exec pulls a task out of the queue and starts a worker goroutine to
 // process the task.
 func (p *processor) exec() {
-	qnames := p.queues()
-	msg, err := p.broker.Dequeue(qnames...)
-	switch {
-	case err == rdb.ErrNoProcessableTask:
-		p.logger.Debug("All queues are empty")
-		// Queues are empty, this is a normal behavior.
-		// Sleep to avoid slamming redis and let scheduler move tasks into queues.
-		// Note: We are not using blocking pop operation and polling queues instead.
-		// This adds significant load to redis.
-		time.Sleep(time.Second)
-		return
-	case err != nil:
-		if p.errLogLimiter.Allow() {
-			p.logger.Errorf("Dequeue error: %v", err)
-		}
-		return
-	}
-
 	select {
-	case <-p.abort:
-		// shutdown is starting, return immediately after requeuing the message.
-		p.requeue(msg)
+	case <-p.quit:
 		return
 	case p.sema <- struct{}{}: // acquire token
+		qnames := p.queues()
+		msg, deadline, err := p.broker.Dequeue(qnames...)
+		switch {
+		case err == rdb.ErrNoProcessableTask:
+			p.logger.Debug("All queues are empty")
+			// Queues are empty, this is a normal behavior.
+			// Sleep to avoid slamming redis and let scheduler move tasks into queues.
+			// Note: We are not using blocking pop operation and polling queues instead.
+			// This adds significant load to redis.
+			time.Sleep(time.Second)
+			<-p.sema // release token
+			return
+		case err != nil:
+			if p.errLogLimiter.Allow() {
+				p.logger.Errorf("Dequeue error: %v", err)
+			}
+			<-p.sema // release token
+			return
+		}
+
 		p.starting <- msg
 		go func() {
 			defer func() {
@@ -189,7 +184,7 @@ func (p *processor) exec() {
 				<-p.sema // release token
 			}()
 
-			ctx, cancel := createContext(msg)
+			ctx, cancel := createContext(msg, deadline)
 			p.cancelations.Add(msg.ID.String(), cancel)
 			defer func() {
 				cancel()
@@ -201,11 +196,15 @@ func (p *processor) exec() {
 			go func() { resCh <- perform(ctx, task, p.handler) }()
 
 			select {
-			case <-p.quit:
+			case <-p.abort:
 				// time is up, push the message back to queue and quit this worker goroutine.
 				p.logger.Warnf("Quitting worker. task id=%s", msg.ID)
 				p.requeue(msg)
 				return
+			case <-ctx.Done():
+				p.logger.Debugf("Retrying task. task id=%s", msg.ID) // TODO: Improve this log message and above
+				p.retryOrKill(ctx, msg, ctx.Err())
+				return
 			case resErr := <-resCh:
 				// Note: One of three things should happen.
 				// 1) Done  -> Removes the message from InProgress
@@ -215,14 +214,10 @@ func (p *processor) exec() {
 					if p.errHandler != nil {
 						p.errHandler.HandleError(task, resErr, msg.Retried, msg.Retry)
 					}
-					if msg.Retried >= msg.Retry {
-						p.kill(msg, resErr)
-					} else {
-						p.retry(msg, resErr)
-					}
+					p.retryOrKill(ctx, msg, resErr)
 					return
 				}
-				p.markAsDone(msg)
+				p.markAsDone(ctx, msg)
 			}
 		}()
 	}
@@ -237,47 +232,70 @@ func (p *processor) requeue(msg *base.TaskMessage) {
 	}
 }
 
-func (p *processor) markAsDone(msg *base.TaskMessage) {
+func (p *processor) markAsDone(ctx context.Context, msg *base.TaskMessage) {
 	err := p.broker.Done(msg)
 	if err != nil {
 		errMsg := fmt.Sprintf("Could not remove task id=%s type=%q from %q err: %+v", msg.ID, msg.Type, base.InProgressQueue, err)
+		deadline, ok := ctx.Deadline()
+		if !ok {
+			panic("asynq: internal error: missing deadline in context")
+		}
 		p.logger.Warnf("%s; Will retry syncing", errMsg)
 		p.syncRequestCh <- &syncRequest{
 			fn: func() error {
 				return p.broker.Done(msg)
 			},
-			errMsg: errMsg,
+			errMsg:   errMsg,
+			deadline: deadline,
 		}
 	}
 }
 
-func (p *processor) retry(msg *base.TaskMessage, e error) {
+func (p *processor) retryOrKill(ctx context.Context, msg *base.TaskMessage, err error) {
+	if msg.Retried >= msg.Retry {
+		p.kill(ctx, msg, err)
+	} else {
+		p.retry(ctx, msg, err)
+	}
+}
+
+func (p *processor) retry(ctx context.Context, msg *base.TaskMessage, e error) {
 	d := p.retryDelayFunc(msg.Retried, e, NewTask(msg.Type, msg.Payload))
 	retryAt := time.Now().Add(d)
 	err := p.broker.Retry(msg, retryAt, e.Error())
 	if err != nil {
 		errMsg := fmt.Sprintf("Could not move task id=%s from %q to %q", msg.ID, base.InProgressQueue, base.RetryQueue)
+		deadline, ok := ctx.Deadline()
+		if !ok {
+			panic("asynq: internal error: missing deadline in context")
+		}
 		p.logger.Warnf("%s; Will retry syncing", errMsg)
 		p.syncRequestCh <- &syncRequest{
 			fn: func() error {
 				return p.broker.Retry(msg, retryAt, e.Error())
 			},
-			errMsg: errMsg,
+			errMsg:   errMsg,
+			deadline: deadline,
 		}
 	}
 }
 
-func (p *processor) kill(msg *base.TaskMessage, e error) {
+func (p *processor) kill(ctx context.Context, msg *base.TaskMessage, e error) {
 	p.logger.Warnf("Retry exhausted for task id=%s", msg.ID)
 	err := p.broker.Kill(msg, e.Error())
 	if err != nil {
 		errMsg := fmt.Sprintf("Could not move task id=%s from %q to %q", msg.ID, base.InProgressQueue, base.DeadQueue)
+		deadline, ok := ctx.Deadline()
+		if !ok {
+			panic("asynq: internal error: missing deadline in context")
+		}
 		p.logger.Warnf("%s; Will retry syncing", errMsg)
 		p.syncRequestCh <- &syncRequest{
 			fn: func() error {
 				return p.broker.Kill(msg, e.Error())
 			},
-			errMsg: errMsg,
+			errMsg:   errMsg,
+			deadline: deadline,
 		}
 	}
 }
diff --git a/recoverer.go b/recoverer.go
new file mode 100644
index 0000000..7fde8a4
--- /dev/null
+++ b/recoverer.go
@@ -0,0 +1,96 @@
+// Copyright 2020 Kentaro Hibino. All rights reserved.
+// Use of this source code is governed by a MIT license
+// that can be found in the LICENSE file.
+
+package asynq
+
+import (
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/hibiken/asynq/internal/base"
+	"github.com/hibiken/asynq/internal/log"
+)
+
+type recoverer struct {
+	logger         *log.Logger
+	broker         base.Broker
+	retryDelayFunc retryDelayFunc
+
+	// channel to communicate back to the long running "recoverer" goroutine.
+	done chan struct{}
+
+	// poll interval.
+	interval time.Duration
+}
+
+type recovererParams struct {
+	logger         *log.Logger
+	broker         base.Broker
+	interval       time.Duration
+	retryDelayFunc retryDelayFunc
+}
+
+func newRecoverer(params recovererParams) *recoverer {
+	return &recoverer{
+		logger:         params.logger,
+		broker:         params.broker,
+		done:           make(chan struct{}),
+		interval:       params.interval,
+		retryDelayFunc: params.retryDelayFunc,
+	}
+}
+
+func (r *recoverer) terminate() {
+	r.logger.Debug("Recoverer shutting down...")
+	// Signal the recoverer goroutine to stop polling.
+	r.done <- struct{}{}
+}
+
+func (r *recoverer) start(wg *sync.WaitGroup) {
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		timer := time.NewTimer(r.interval)
+		for {
+			select {
+			case <-r.done:
+				r.logger.Debug("Recoverer done")
+				timer.Stop()
+				return
+			case <-timer.C:
+				// Get all tasks which have expired 30 seconds ago or earlier.
+				deadline := time.Now().Add(-30 * time.Second)
+				msgs, err := r.broker.ListDeadlineExceeded(deadline)
+				if err != nil {
+					r.logger.Warn("recoverer: could not list deadline exceeded tasks")
+					continue
+				}
+				const errMsg = "deadline exceeded" // TODO: better error message
+				for _, msg := range msgs {
+					if msg.Retried >= msg.Retry {
+						r.kill(msg, errMsg)
+					} else {
+						r.retry(msg, errMsg)
+					}
+				}
+
+			}
+		}
+	}()
+}
+
+func (r *recoverer) retry(msg *base.TaskMessage, errMsg string) {
+	delay := r.retryDelayFunc(msg.Retried, fmt.Errorf(errMsg), NewTask(msg.Type, msg.Payload))
+	retryAt := time.Now().Add(delay)
+	if err := r.broker.Retry(msg, retryAt, errMsg); err != nil {
+		r.logger.Warnf("recoverer: could not retry deadline exceeded task: %v", err)
+	}
+}
+
+func (r *recoverer) kill(msg *base.TaskMessage, errMsg string) {
+	if err := r.broker.Kill(msg, errMsg); err != nil {
+		r.logger.Warnf("recoverer: could not move task to dead queue: %v", err)
+	}
+}
diff --git a/server.go b/server.go
index 896c03c..95e7ee3 100644
--- a/server.go
+++ b/server.go
@@ -46,6 +46,7 @@ type Server struct {
 	syncer      *syncer
 	heartbeater *heartbeater
 	subscriber  *subscriber
+	recoverer   *recoverer
 }
 
 // Config specifies the server's background-task processing behavior.
@@ -329,6 +330,12 @@ func NewServer(r RedisConnOpt, cfg Config) *Server {
 		starting:        starting,
 		finished:        finished,
 	})
+	recoverer := newRecoverer(recovererParams{
+		logger:         logger,
+		broker:         rdb,
+		retryDelayFunc: delayFunc,
+		interval:       1 * time.Minute,
+	})
 	return &Server{
 		logger:      logger,
 		broker:      rdb,
@@ -338,6 +345,7 @@ func NewServer(r RedisConnOpt, cfg Config) *Server {
 		syncer:      syncer,
 		heartbeater: heartbeater,
 		subscriber:  subscriber,
+		recoverer:   recoverer,
 	}
 }
 
@@ -407,6 +415,7 @@ func (srv *Server) Start(handler Handler) error {
 	srv.heartbeater.start(&srv.wg)
 	srv.subscriber.start(&srv.wg)
 	srv.syncer.start(&srv.wg)
+	srv.recoverer.start(&srv.wg)
 	srv.scheduler.start(&srv.wg)
 	srv.processor.start(&srv.wg)
 	return nil
@@ -430,6 +439,7 @@ func (srv *Server) Stop() {
 	// processor -> heartbeater (via starting, finished channels)
 	srv.scheduler.terminate()
 	srv.processor.terminate()
+	srv.recoverer.terminate()
 	srv.syncer.terminate()
 	srv.subscriber.terminate()
 	srv.heartbeater.terminate()
diff --git a/syncer.go b/syncer.go
index c5ca396..d108a7a 100644
--- a/syncer.go
+++ b/syncer.go
@@ -26,8 +26,9 @@ type syncer struct {
 }
 
 type syncRequest struct {
-	fn     func() error // sync operation
-	errMsg string       // error message
+	fn       func() error // sync operation
+	errMsg   string       // error message
+	deadline time.Time    // request should be dropped if deadline has been exceeded
 }
 
 type syncerParams struct {
@@ -72,6 +73,9 @@ func (s *syncer) start(wg *sync.WaitGroup) {
 			case <-time.After(s.interval):
 				var temp []*syncRequest
 				for _, req := range requests {
+					if req.deadline.Before(time.Now()) {
+						continue // drop stale request
+					}
 					if err := req.fn(); err != nil {
 						temp = append(temp, req)
 					}
diff --git a/tools/asynq/cmd/migrate.go b/tools/asynq/cmd/migrate.go
new file mode 100644
index 0000000..cf883cf
--- /dev/null
+++ b/tools/asynq/cmd/migrate.go
@@ -0,0 +1,214 @@
+// Copyright 2020 Kentaro Hibino. All rights reserved.
+// Use of this source code is governed by a MIT license
+// that can be found in the LICENSE file.
+
+package cmd
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/go-redis/redis/v7"
+	"github.com/hibiken/asynq/internal/base"
+	"github.com/rs/xid"
+	"github.com/spf13/cast"
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+)
+
+const version = "v0.10.0" // TODO: Define this version string in a more canonical place.
+
+// migrateCmd represents the migrate command
+var migrateCmd = &cobra.Command{
+	Use:   "migrate",
+	Short: fmt.Sprintf("Migrate all tasks to be compatible with asynq@%s", version),
+	Long:  fmt.Sprintf("Migrate (asynq migrate) will convert all tasks in redis to be compatible with asynq@%s.", version),
+	Run:   migrate,
+}
+
+func init() {
+	rootCmd.AddCommand(migrateCmd)
+}
+
+func migrate(cmd *cobra.Command, args []string) {
+	c := redis.NewClient(&redis.Options{
+		Addr:     viper.GetString("uri"),
+		DB:       viper.GetInt("db"),
+		Password: viper.GetString("password"),
+	})
+
+	lists := []string{base.InProgressQueue}
+	allQueues, err := c.SMembers(base.AllQueues).Result()
+	if err != nil {
+		fmt.Printf("error: could not read all queues: %v", err)
+		os.Exit(1)
+	}
+	lists = append(lists, allQueues...)
+	for _, key := range lists {
+		if err := migrateList(c, key); err != nil {
+			fmt.Printf("error: %v", err)
+			os.Exit(1)
+		}
+	}
+
+	zsets := []string{base.ScheduledQueue, base.RetryQueue, base.DeadQueue}
+	for _, key := range zsets {
+		if err := migrateZSet(c, key); err != nil {
+			fmt.Printf("error: %v", err)
+			os.Exit(1)
+		}
+	}
+}
+
+type oldTaskMessage struct {
+	// Unchanged
+	Type      string
+	Payload   map[string]interface{}
+	ID        xid.ID
+	Queue     string
+	Retry     int
+	Retried   int
+	ErrorMsg  string
+	UniqueKey string
+
+	// Following fields have changed.
+
+	// Deadline specifies the deadline for the task.
+	// Task won't be processed if it exceeded its deadline.
+	// The string shoulbe be in RFC3339 format.
+	//
+	// time.Time's zero value means no deadline.
+	Timeout string
+
+	// Deadline specifies the deadline for the task.
+	// Task won't be processed if it exceeded its deadline.
+	// The string shoulbe be in RFC3339 format.
+	//
+	// time.Time's zero value means no deadline.
+	Deadline string
+}
+
+var defaultTimeout = 30 * time.Minute
+
+func convertMessage(old *oldTaskMessage) (*base.TaskMessage, error) {
+	timeout, err := time.ParseDuration(old.Timeout)
+	if err != nil {
+		return nil, fmt.Errorf("could not parse Timeout field of %+v", old)
+	}
+	deadline, err := time.Parse(time.RFC3339, old.Deadline)
+	if err != nil {
+		return nil, fmt.Errorf("could not parse Deadline field of %+v", old)
+	}
+	if timeout == 0 && deadline.IsZero() {
+		timeout = defaultTimeout
+	}
+	if deadline.IsZero() {
+		// Zero value used to be time.Time{},
+		// in the new schema zero value is represented by
+		// zero in Unix time.
+		deadline = time.Unix(0, 0)
+	}
+	return &base.TaskMessage{
+		Type:      old.Type,
+		Payload:   old.Payload,
+		ID:        old.ID,
+		Queue:     old.Queue,
+		Retry:     old.Retry,
+		Retried:   old.Retried,
+		ErrorMsg:  old.ErrorMsg,
+		UniqueKey: old.UniqueKey,
+		Timeout:   int64(timeout.Seconds()),
+		Deadline:  deadline.Unix(),
+	}, nil
+}
+
+func deserialize(s string) (*base.TaskMessage, error) {
+	// Try deserializing as old message.
+	d := json.NewDecoder(strings.NewReader(s))
+	d.UseNumber()
+	var old *oldTaskMessage
+	if err := d.Decode(&old); err != nil {
+		// Try deserializing as new message.
+		d = json.NewDecoder(strings.NewReader(s))
+		d.UseNumber()
+		var msg *base.TaskMessage
+		if err := d.Decode(&msg); err != nil {
+			return nil, fmt.Errorf("could not deserialize %s into task message: %v", s, err)
+		}
+		return msg, nil
+	}
+	return convertMessage(old)
+}
+
+func migrateZSet(c *redis.Client, key string) error {
+	if c.Exists(key).Val() == 0 {
+		// skip if key doesn't exist.
+		return nil
+	}
+	res, err := c.ZRangeWithScores(key, 0, -1).Result()
+	if err != nil {
+		return err
+	}
+	var msgs []*redis.Z
+	for _, z := range res {
+		s, err := cast.ToStringE(z.Member)
+		if err != nil {
+			return fmt.Errorf("could not cast to string: %v", err)
+		}
+		msg, err := deserialize(s)
+		if err != nil {
+			return err
+		}
+		encoded, err := base.EncodeMessage(msg)
+		if err != nil {
+			return fmt.Errorf("could not encode message from %q: %v", key, err)
+		}
+		msgs = append(msgs, &redis.Z{Score: z.Score, Member: encoded})
+	}
+	if err := c.Rename(key, key+":backup").Err(); err != nil {
+		return fmt.Errorf("could not rename key %q: %v", key, err)
+	}
+	if err := c.ZAdd(key, msgs...).Err(); err != nil {
+		return fmt.Errorf("could not write new messages to %q: %v", key, err)
+	}
+	if err := c.Del(key + ":backup").Err(); err != nil {
+		return fmt.Errorf("could not delete back up key %q: %v", key+":backup", err)
+	}
+	return nil
+}
+
+func migrateList(c *redis.Client, key string) error {
+	if c.Exists(key).Val() == 0 {
+		// skip if key doesn't exist.
+		return nil
+	}
+	res, err := c.LRange(key, 0, -1).Result()
+	if err != nil {
+		return err
+	}
+	var msgs []interface{}
+	for _, s := range res {
+		msg, err := deserialize(s)
+		if err != nil {
+			return err
+		}
+		encoded, err := base.EncodeMessage(msg)
+		if err != nil {
+			return fmt.Errorf("could not encode message from %q: %v", key, err)
+		}
+		msgs = append(msgs, encoded)
+	}
+	if err := c.Rename(key, key+":backup").Err(); err != nil {
+		return fmt.Errorf("could not rename key %q: %v", key, err)
+	}
+	if err := c.LPush(key, msgs...).Err(); err != nil {
+		return fmt.Errorf("could not write new messages to %q: %v", key, err)
+	}
+	if err := c.Del(key + ":backup").Err(); err != nil {
+		return fmt.Errorf("could not delete back up key %q: %v", key+":backup", err)
+	}
+	return nil
+}
diff --git a/tools/go.mod b/tools/go.mod
index 9685df1..97bfab6 100644
--- a/tools/go.mod
+++ b/tools/go.mod
@@ -7,6 +7,7 @@ require (
 	github.com/hibiken/asynq v0.4.0
 	github.com/mitchellh/go-homedir v1.1.0
 	github.com/rs/xid v1.2.1
+	github.com/spf13/cast v1.3.1
 	github.com/spf13/cobra v0.0.5
 	github.com/spf13/viper v1.6.2
 )


Test patch:
diff --git a/client_test.go b/client_test.go
index 348d056..4fd3d65 100644
--- a/client_test.go
+++ b/client_test.go
@@ -15,11 +15,6 @@ import (
 	"github.com/hibiken/asynq/internal/base"
 )
 
-var (
-	noTimeout  = time.Duration(0).String()
-	noDeadline = time.Time{}.Format(time.RFC3339)
-)
-
 func TestClientEnqueueAt(t *testing.T) {
 	r := setup(t)
 	client := NewClient(RedisClientOpt{
@@ -54,8 +49,8 @@ func TestClientEnqueueAt(t *testing.T) {
 						Payload:  task.Payload.data,
 						Retry:    defaultMaxRetry,
 						Queue:    "default",
-						Timeout:  noTimeout,
-						Deadline: noDeadline,
+						Timeout:  int64(defaultTimeout.Seconds()),
+						Deadline: noDeadline.Unix(),
 					},
 				},
 			},
@@ -74,8 +69,8 @@ func TestClientEnqueueAt(t *testing.T) {
 						Payload:  task.Payload.data,
 						Retry:    defaultMaxRetry,
 						Queue:    "default",
-						Timeout:  noTimeout,
-						Deadline: noDeadline,
+						Timeout:  int64(defaultTimeout.Seconds()),
+						Deadline: noDeadline.Unix(),
 					},
 					Score: float64(oneHourLater.Unix()),
 				},
@@ -134,8 +129,8 @@ func TestClientEnqueue(t *testing.T) {
 						Payload:  task.Payload.data,
 						Retry:    3,
 						Queue:    "default",
-						Timeout:  noTimeout,
-						Deadline: noDeadline,
+						Timeout:  int64(defaultTimeout.Seconds()),
+						Deadline: noDeadline.Unix(),
 					},
 				},
 			},
@@ -153,8 +148,8 @@ func TestClientEnqueue(t *testing.T) {
 						Payload:  task.Payload.data,
 						Retry:    0, // Retry count should be set to zero
 						Queue:    "default",
-						Timeout:  noTimeout,
-						Deadline: noDeadline,
+						Timeout:  int64(defaultTimeout.Seconds()),
+						Deadline: noDeadline.Unix(),
 					},
 				},
 			},
@@ -173,8 +168,8 @@ func TestClientEnqueue(t *testing.T) {
 						Payload:  task.Payload.data,
 						Retry:    10, // Last option takes precedence
 						Queue:    "default",
-						Timeout:  noTimeout,
-						Deadline: noDeadline,
+						Timeout:  int64(defaultTimeout.Seconds()),
+						Deadline: noDeadline.Unix(),
 					},
 				},
 			},
@@ -192,8 +187,8 @@ func TestClientEnqueue(t *testing.T) {
 						Payload:  task.Payload.data,
 						Retry:    defaultMaxRetry,
 						Queue:    "custom",
-						Timeout:  noTimeout,
-						Deadline: noDeadline,
+						Timeout:  int64(defaultTimeout.Seconds()),
+						Deadline: noDeadline.Unix(),
 					},
 				},
 			},
@@ -211,8 +206,8 @@ func TestClientEnqueue(t *testing.T) {
 						Payload:  task.Payload.data,
 						Retry:    defaultMaxRetry,
 						Queue:    "high",
-						Timeout:  noTimeout,
-						Deadline: noDeadline,
+						Timeout:  int64(defaultTimeout.Seconds()),
+						Deadline: noDeadline.Unix(),
 					},
 				},
 			},
@@ -230,8 +225,8 @@ func TestClientEnqueue(t *testing.T) {
 						Payload:  task.Payload.data,
 						Retry:    defaultMaxRetry,
 						Queue:    "default",
-						Timeout:  (20 * time.Second).String(),
-						Deadline: noDeadline,
+						Timeout:  20,
+						Deadline: noDeadline.Unix(),
 					},
 				},
 			},
@@ -249,8 +244,28 @@ func TestClientEnqueue(t *testing.T) {
 						Payload:  task.Payload.data,
 						Retry:    defaultMaxRetry,
 						Queue:    "default",
-						Timeout:  noTimeout,
-						Deadline: time.Date(2020, time.June, 24, 0, 0, 0, 0, time.UTC).Format(time.RFC3339),
+						Timeout:  int64(noTimeout.Seconds()),
+						Deadline: time.Date(2020, time.June, 24, 0, 0, 0, 0, time.UTC).Unix(),
+					},
+				},
+			},
+		},
+		{
+			desc: "With both deadline and timeout options",
+			task: task,
+			opts: []Option{
+				Timeout(20 * time.Second),
+				Deadline(time.Date(2020, time.June, 24, 0, 0, 0, 0, time.UTC)),
+			},
+			wantEnqueued: map[string][]*base.TaskMessage{
+				"default": {
+					{
+						Type:     task.Type,
+						Payload:  task.Payload.data,
+						Retry:    defaultMaxRetry,
+						Queue:    "default",
+						Timeout:  20,
+						Deadline: time.Date(2020, time.June, 24, 0, 0, 0, 0, time.UTC).Unix(),
 					},
 				},
 			},
@@ -305,8 +320,8 @@ func TestClientEnqueueIn(t *testing.T) {
 						Payload:  task.Payload.data,
 						Retry:    defaultMaxRetry,
 						Queue:    "default",
-						Timeout:  noTimeout,
-						Deadline: noDeadline,
+						Timeout:  int64(defaultTimeout.Seconds()),
+						Deadline: noDeadline.Unix(),
 					},
 					Score: float64(time.Now().Add(time.Hour).Unix()),
 				},
@@ -324,8 +339,8 @@ func TestClientEnqueueIn(t *testing.T) {
 						Payload:  task.Payload.data,
 						Retry:    defaultMaxRetry,
 						Queue:    "default",
-						Timeout:  noTimeout,
-						Deadline: noDeadline,
+						Timeout:  int64(defaultTimeout.Seconds()),
+						Deadline: noDeadline.Unix(),
 					},
 				},
 			},
@@ -378,8 +393,8 @@ func TestClientDefaultOptions(t *testing.T) {
 				Payload:  nil,
 				Retry:    defaultMaxRetry,
 				Queue:    "feed",
-				Timeout:  noTimeout,
-				Deadline: noDeadline,
+				Timeout:  int64(defaultTimeout.Seconds()),
+				Deadline: noDeadline.Unix(),
 			},
 		},
 		{
@@ -393,8 +408,8 @@ func TestClientDefaultOptions(t *testing.T) {
 				Payload:  nil,
 				Retry:    5,
 				Queue:    "feed",
-				Timeout:  noTimeout,
-				Deadline: noDeadline,
+				Timeout:  int64(defaultTimeout.Seconds()),
+				Deadline: noDeadline.Unix(),
 			},
 		},
 		{
@@ -408,8 +423,8 @@ func TestClientDefaultOptions(t *testing.T) {
 				Payload:  nil,
 				Retry:    5,
 				Queue:    "critical",
-				Timeout:  noTimeout,
-				Deadline: noDeadline,
+				Timeout:  int64(defaultTimeout.Seconds()),
+				Deadline: noDeadline.Unix(),
 			},
 		},
 	}
diff --git a/context_test.go b/context_test.go
index 1a5eff6..261def5 100644
--- a/context_test.go
+++ b/context_test.go
@@ -10,51 +10,38 @@ import (
 	"time"
 
 	"github.com/google/go-cmp/cmp"
-	"github.com/google/go-cmp/cmp/cmpopts"
 	"github.com/hibiken/asynq/internal/base"
 	"github.com/rs/xid"
 )
 
-func TestCreateContextWithTimeRestrictions(t *testing.T) {
-	var (
-		noTimeout  = time.Duration(0)
-		noDeadline = time.Time{}
-	)
-
+func TestCreateContextWithFutureDeadline(t *testing.T) {
 	tests := []struct {
-		desc         string
-		timeout      time.Duration
-		deadline     time.Time
-		wantDeadline time.Time
+		deadline time.Time
 	}{
-		{"only with timeout", 10 * time.Second, noDeadline, time.Now().Add(10 * time.Second)},
-		{"only with deadline", noTimeout, time.Now().Add(time.Hour), time.Now().Add(time.Hour)},
-		{"with timeout and deadline (timeout < deadline)", 10 * time.Second, time.Now().Add(time.Hour), time.Now().Add(10 * time.Second)},
-		{"with timeout and deadline (timeout > deadline)", 10 * time.Minute, time.Now().Add(30 * time.Second), time.Now().Add(30 * time.Second)},
+		{time.Now().Add(time.Hour)},
 	}
 
 	for _, tc := range tests {
 		msg := &base.TaskMessage{
-			Type:     "something",
-			ID:       xid.New(),
-			Timeout:  tc.timeout.String(),
-			Deadline: tc.deadline.Format(time.RFC3339),
+			Type:    "something",
+			ID:      xid.New(),
+			Payload: nil,
 		}
 
-		ctx, cancel := createContext(msg)
+		ctx, cancel := createContext(msg, tc.deadline)
 
 		select {
 		case x := <-ctx.Done():
-			t.Errorf("%s: <-ctx.Done() == %v, want nothing (it should block)", tc.desc, x)
+			t.Errorf("<-ctx.Done() == %v, want nothing (it should block)", x)
 		default:
 		}
 
 		got, ok := ctx.Deadline()
 		if !ok {
-			t.Errorf("%s: ctx.Deadline() returned false, want deadline to be set", tc.desc)
+			t.Errorf("ctx.Deadline() returned false, want deadline to be set")
 		}
-		if !cmp.Equal(tc.wantDeadline, got, cmpopts.EquateApproxTime(time.Second)) {
-			t.Errorf("%s: ctx.Deadline() returned %v, want %v", tc.desc, got, tc.wantDeadline)
+		if !cmp.Equal(tc.deadline, got) {
+			t.Errorf("ctx.Deadline() returned %v, want %v", got, tc.deadline)
 		}
 
 		cancel()
@@ -67,33 +54,36 @@ func TestCreateContextWithTimeRestrictions(t *testing.T) {
 	}
 }
 
-func TestCreateContextWithoutTimeRestrictions(t *testing.T) {
-	msg := &base.TaskMessage{
-		Type:     "something",
-		ID:       xid.New(),
-		Timeout:  time.Duration(0).String(),        // zero value to indicate no timeout
-		Deadline: time.Time{}.Format(time.RFC3339), // zero value to indicate no deadline
+func TestCreateContextWithPastDeadline(t *testing.T) {
+	tests := []struct {
+		deadline time.Time
+	}{
+		{time.Now().Add(-2 * time.Hour)},
 	}
 
-	ctx, cancel := createContext(msg)
-
-	select {
-	case x := <-ctx.Done():
-		t.Errorf("<-ctx.Done() == %v, want nothing (it should block)", x)
-	default:
-	}
+	for _, tc := range tests {
+		msg := &base.TaskMessage{
+			Type:    "something",
+			ID:      xid.New(),
+			Payload: nil,
+		}
 
-	_, ok := ctx.Deadline()
-	if ok {
-		t.Error("ctx.Deadline() returned true, want deadline to not be set")
-	}
+		ctx, cancel := createContext(msg, tc.deadline)
+		defer cancel()
 
-	cancel()
+		select {
+		case <-ctx.Done():
+		default:
+			t.Errorf("ctx.Done() blocked, want it to be non-blocking")
+		}
 
-	select {
-	case <-ctx.Done():
-	default:
-		t.Error("ctx.Done() blocked, want it to be non-blocking")
+		got, ok := ctx.Deadline()
+		if !ok {
+			t.Errorf("ctx.Deadline() returned false, want deadline to be set")
+		}
+		if !cmp.Equal(tc.deadline, got) {
+			t.Errorf("ctx.Deadline() returned %v, want %v", got, tc.deadline)
+		}
 	}
 }
 
@@ -102,12 +92,13 @@ func TestGetTaskMetadataFromContext(t *testing.T) {
 		desc string
 		msg  *base.TaskMessage
 	}{
-		{"with zero retried message", &base.TaskMessage{Type: "something", ID: xid.New(), Retry: 25, Retried: 0}},
-		{"with non-zero retried message", &base.TaskMessage{Type: "something", ID: xid.New(), Retry: 10, Retried: 5}},
+		{"with zero retried message", &base.TaskMessage{Type: "something", ID: xid.New(), Retry: 25, Retried: 0, Timeout: 1800}},
+		{"with non-zero retried message", &base.TaskMessage{Type: "something", ID: xid.New(), Retry: 10, Retried: 5, Timeout: 1800}},
 	}
 
 	for _, tc := range tests {
-		ctx, _ := createContext(tc.msg)
+		ctx, cancel := createContext(tc.msg, time.Now().Add(30*time.Minute))
+		defer cancel()
 
 		id, ok := GetTaskID(ctx)
 		if !ok {
diff --git a/internal/asynqtest/asynqtest.go b/internal/asynqtest/asynqtest.go
index a713deb..4df8d2b 100644
--- a/internal/asynqtest/asynqtest.go
+++ b/internal/asynqtest/asynqtest.go
@@ -75,11 +75,13 @@ var IgnoreIDOpt = cmpopts.IgnoreFields(base.TaskMessage{}, "ID")
 // NewTaskMessage returns a new instance of TaskMessage given a task type and payload.
 func NewTaskMessage(taskType string, payload map[string]interface{}) *base.TaskMessage {
 	return &base.TaskMessage{
-		ID:      xid.New(),
-		Type:    taskType,
-		Queue:   base.DefaultQueueName,
-		Retry:   25,
-		Payload: payload,
+		ID:       xid.New(),
+		Type:     taskType,
+		Queue:    base.DefaultQueueName,
+		Retry:    25,
+		Payload:  payload,
+		Timeout:  1800, // default timeout of 30 mins
+		Deadline: 0,    // no deadline
 	}
 }
 
@@ -95,6 +97,20 @@ func NewTaskMessageWithQueue(taskType string, payload map[string]interface{}, qn
 	}
 }
 
+// TaskMessageAfterRetry returns an updated copy of t after retry.
+// It increments retry count and sets the error message.
+func TaskMessageAfterRetry(t base.TaskMessage, errMsg string) *base.TaskMessage {
+	t.Retried = t.Retried + 1
+	t.ErrorMsg = errMsg
+	return &t
+}
+
+// TaskMessageWithError returns an updated copy of t with the given error message.
+func TaskMessageWithError(t base.TaskMessage, errMsg string) *base.TaskMessage {
+	t.ErrorMsg = errMsg
+	return &t
+}
+
 // MustMarshal marshals given task message and returns a json string.
 // Calling test will fail if marshaling errors out.
 func MustMarshal(tb testing.TB, msg *base.TaskMessage) string {
@@ -185,6 +201,12 @@ func SeedDeadQueue(tb testing.TB, r *redis.Client, entries []ZSetEntry) {
 	seedRedisZSet(tb, r, base.DeadQueue, entries)
 }
 
+// SeedDeadlines initializes the deadlines set with the given entries.
+func SeedDeadlines(tb testing.TB, r *redis.Client, entries []ZSetEntry) {
+	tb.Helper()
+	seedRedisZSet(tb, r, base.KeyDeadlines, entries)
+}
+
 func seedRedisList(tb testing.TB, c *redis.Client, key string, msgs []*base.TaskMessage) {
 	data := MustMarshalSlice(tb, msgs)
 	for _, s := range data {
@@ -257,6 +279,12 @@ func GetDeadEntries(tb testing.TB, r *redis.Client) []ZSetEntry {
 	return getZSetEntries(tb, r, base.DeadQueue)
 }
 
+// GetDeadlinesEntries returns all task messages and its score in the deadlines set.
+func GetDeadlinesEntries(tb testing.TB, r *redis.Client) []ZSetEntry {
+	tb.Helper()
+	return getZSetEntries(tb, r, base.KeyDeadlines)
+}
+
 func getListMessages(tb testing.TB, r *redis.Client, list string) []*base.TaskMessage {
 	data := r.LRange(list, 0, -1).Val()
 	return MustUnmarshalSlice(tb, data)
diff --git a/internal/base/base_test.go b/internal/base/base_test.go
index 700ae53..f63688a 100644
--- a/internal/base/base_test.go
+++ b/internal/base/base_test.go
@@ -115,22 +115,24 @@ func TestMessageEncoding(t *testing.T) {
 	}{
 		{
 			in: &TaskMessage{
-				Type:    "task1",
-				Payload: map[string]interface{}{"a": 1, "b": "hello!", "c": true},
-				ID:      id,
-				Queue:   "default",
-				Retry:   10,
-				Retried: 0,
-				Timeout: "0",
+				Type:     "task1",
+				Payload:  map[string]interface{}{"a": 1, "b": "hello!", "c": true},
+				ID:       id,
+				Queue:    "default",
+				Retry:    10,
+				Retried:  0,
+				Timeout:  1800,
+				Deadline: 1692311100,
 			},
 			out: &TaskMessage{
-				Type:    "task1",
-				Payload: map[string]interface{}{"a": json.Number("1"), "b": "hello!", "c": true},
-				ID:      id,
-				Queue:   "default",
-				Retry:   10,
-				Retried: 0,
-				Timeout: "0",
+				Type:     "task1",
+				Payload:  map[string]interface{}{"a": json.Number("1"), "b": "hello!", "c": true},
+				ID:       id,
+				Queue:    "default",
+				Retry:    10,
+				Retried:  0,
+				Timeout:  1800,
+				Deadline: 1692311100,
 			},
 		},
 	}
@@ -220,9 +222,4 @@ func TestCancelationsConcurrentAccess(t *testing.T) {
 	if ok {
 		t.Errorf("(*Cancelations).Get(%q) = _, true, want <nil>, false", key2)
 	}
-
-	funcs := c.GetAll()
-	if len(funcs) != 2 {
-		t.Errorf("(*Cancelations).GetAll() returns %d functions, want 2", len(funcs))
-	}
 }
diff --git a/internal/rdb/rdb_test.go b/internal/rdb/rdb_test.go
index ca3389d..e7a95e8 100644
--- a/internal/rdb/rdb_test.go
+++ b/internal/rdb/rdb_test.go
@@ -114,41 +114,74 @@ func TestEnqueueUnique(t *testing.T) {
 
 func TestDequeue(t *testing.T) {
 	r := setup(t)
-	t1 := h.NewTaskMessage("send_email", map[string]interface{}{"subject": "hello!"})
-	t2 := h.NewTaskMessage("export_csv", nil)
-	t3 := h.NewTaskMessage("reindex", nil)
+	now := time.Now()
+	t1 := &base.TaskMessage{
+		ID:       xid.New(),
+		Type:     "send_email",
+		Payload:  map[string]interface{}{"subject": "hello!"},
+		Timeout:  1800,
+		Deadline: 0,
+	}
+	t1Deadline := now.Unix() + t1.Timeout
+	t2 := &base.TaskMessage{
+		ID:       xid.New(),
+		Type:     "export_csv",
+		Payload:  nil,
+		Timeout:  0,
+		Deadline: 1593021600,
+	}
+	t2Deadline := t2.Deadline
+	t3 := &base.TaskMessage{
+		ID:       xid.New(),
+		Type:     "reindex",
+		Payload:  nil,
+		Timeout:  int64((5 * time.Minute).Seconds()),
+		Deadline: time.Now().Add(10 * time.Minute).Unix(),
+	}
+	t3Deadline := now.Unix() + t3.Timeout // use whichever is earliest
 
 	tests := []struct {
 		enqueued       map[string][]*base.TaskMessage
 		args           []string // list of queues to query
-		want           *base.TaskMessage
+		wantMsg        *base.TaskMessage
+		wantDeadline   time.Time
 		err            error
 		wantEnqueued   map[string][]*base.TaskMessage
 		wantInProgress []*base.TaskMessage
+		wantDeadlines  []h.ZSetEntry
 	}{
 		{
 			enqueued: map[string][]*base.TaskMessage{
 				"default": {t1},
 			},
-			args: []string{"default"},
-			want: t1,
-			err:  nil,
+			args:         []string{"default"},
+			wantMsg:      t1,
+			wantDeadline: time.Unix(t1Deadline, 0),
+			err:          nil,
 			wantEnqueued: map[string][]*base.TaskMessage{
 				"default": {},
 			},
 			wantInProgress: []*base.TaskMessage{t1},
+			wantDeadlines: []h.ZSetEntry{
+				{
+					Msg:   t1,
+					Score: float64(t1Deadline),
+				},
+			},
 		},
 		{
 			enqueued: map[string][]*base.TaskMessage{
 				"default": {},
 			},
-			args: []string{"default"},
-			want: nil,
-			err:  ErrNoProcessableTask,
+			args:         []string{"default"},
+			wantMsg:      nil,
+			wantDeadline: time.Time{},
+			err:          ErrNoProcessableTask,
 			wantEnqueued: map[string][]*base.TaskMessage{
 				"default": {},
 			},
 			wantInProgress: []*base.TaskMessage{},
+			wantDeadlines:  []h.ZSetEntry{},
 		},
 		{
 			enqueued: map[string][]*base.TaskMessage{
@@ -156,31 +189,45 @@ func TestDequeue(t *testing.T) {
 				"critical": {t2},
 				"low":      {t3},
 			},
-			args: []string{"critical", "default", "low"},
-			want: t2,
-			err:  nil,
+			args:         []string{"critical", "default", "low"},
+			wantMsg:      t2,
+			wantDeadline: time.Unix(t2Deadline, 0),
+			err:          nil,
 			wantEnqueued: map[string][]*base.TaskMessage{
 				"default":  {t1},
 				"critical": {},
 				"low":      {t3},
 			},
 			wantInProgress: []*base.TaskMessage{t2},
+			wantDeadlines: []h.ZSetEntry{
+				{
+					Msg:   t2,
+					Score: float64(t2Deadline),
+				},
+			},
 		},
 		{
 			enqueued: map[string][]*base.TaskMessage{
-				"default":  {t1},
+				"default":  {t3},
 				"critical": {},
-				"low":      {t2, t3},
+				"low":      {t2, t1},
 			},
-			args: []string{"critical", "default", "low"},
-			want: t1,
-			err:  nil,
+			args:         []string{"critical", "default", "low"},
+			wantMsg:      t3,
+			wantDeadline: time.Unix(t3Deadline, 0),
+			err:          nil,
 			wantEnqueued: map[string][]*base.TaskMessage{
 				"default":  {},
 				"critical": {},
-				"low":      {t2, t3},
+				"low":      {t2, t1},
+			},
+			wantInProgress: []*base.TaskMessage{t3},
+			wantDeadlines: []h.ZSetEntry{
+				{
+					Msg:   t3,
+					Score: float64(t3Deadline),
+				},
 			},
-			wantInProgress: []*base.TaskMessage{t1},
 		},
 		{
 			enqueued: map[string][]*base.TaskMessage{
@@ -188,15 +235,17 @@ func TestDequeue(t *testing.T) {
 				"critical": {},
 				"low":      {},
 			},
-			args: []string{"critical", "default", "low"},
-			want: nil,
-			err:  ErrNoProcessableTask,
+			args:         []string{"critical", "default", "low"},
+			wantMsg:      nil,
+			wantDeadline: time.Time{},
+			err:          ErrNoProcessableTask,
 			wantEnqueued: map[string][]*base.TaskMessage{
 				"default":  {},
 				"critical": {},
 				"low":      {},
 			},
 			wantInProgress: []*base.TaskMessage{},
+			wantDeadlines:  []h.ZSetEntry{},
 		},
 	}
 
@@ -206,10 +255,20 @@ func TestDequeue(t *testing.T) {
 			h.SeedEnqueuedQueue(t, r.client, msgs, queue)
 		}
 
-		got, err := r.Dequeue(tc.args...)
-		if !cmp.Equal(got, tc.want) || err != tc.err {
-			t.Errorf("(*RDB).Dequeue(%v) = %v, %v; want %v, %v",
-				tc.args, got, err, tc.want, tc.err)
+		gotMsg, gotDeadline, err := r.Dequeue(tc.args...)
+		if err != tc.err {
+			t.Errorf("(*RDB).Dequeue(%v) returned error %v; want %v",
+				tc.args, err, tc.err)
+			continue
+		}
+		if !cmp.Equal(gotMsg, tc.wantMsg) || err != tc.err {
+			t.Errorf("(*RDB).Dequeue(%v) returned message %v; want %v",
+				tc.args, gotMsg, tc.wantMsg)
+			continue
+		}
+		if gotDeadline != tc.wantDeadline {
+			t.Errorf("(*RDB).Dequeue(%v) returned deadline %v; want %v",
+				tc.args, gotDeadline, tc.wantDeadline)
 			continue
 		}
 
@@ -219,11 +278,14 @@ func TestDequeue(t *testing.T) {
 				t.Errorf("mismatch found in %q: (-want,+got):\n%s", base.QueueKey(queue), diff)
 			}
 		}
-
 		gotInProgress := h.GetInProgressMessages(t, r.client)
 		if diff := cmp.Diff(tc.wantInProgress, gotInProgress, h.SortMsgOpt); diff != "" {
 			t.Errorf("mismatch found in %q: (-want,+got):\n%s", base.InProgressQueue, diff)
 		}
+		gotDeadlines := h.GetDeadlinesEntries(t, r.client)
+		if diff := cmp.Diff(tc.wantDeadlines, gotDeadlines, h.SortZSetEntryOpt); diff != "" {
+			t.Errorf("mismatch found in %q: (-want,+got):\n%s", base.KeyDeadlines, diff)
+		}
 	}
 }
 
@@ -236,7 +298,7 @@ func TestDequeueIgnoresPausedQueues(t *testing.T) {
 		paused         []string // list of paused queues
 		enqueued       map[string][]*base.TaskMessage
 		args           []string // list of queues to query
-		want           *base.TaskMessage
+		wantMsg        *base.TaskMessage
 		err            error
 		wantEnqueued   map[string][]*base.TaskMessage
 		wantInProgress []*base.TaskMessage
@@ -247,9 +309,9 @@ func TestDequeueIgnoresPausedQueues(t *testing.T) {
 				"default":  {t1},
 				"critical": {t2},
 			},
-			args: []string{"default", "critical"},
-			want: t2,
-			err:  nil,
+			args:    []string{"default", "critical"},
+			wantMsg: t2,
+			err:     nil,
 			wantEnqueued: map[string][]*base.TaskMessage{
 				"default":  {t1},
 				"critical": {},
@@ -261,9 +323,9 @@ func TestDequeueIgnoresPausedQueues(t *testing.T) {
 			enqueued: map[string][]*base.TaskMessage{
 				"default": {t1},
 			},
-			args: []string{"default"},
-			want: nil,
-			err:  ErrNoProcessableTask,
+			args:    []string{"default"},
+			wantMsg: nil,
+			err:     ErrNoProcessableTask,
 			wantEnqueued: map[string][]*base.TaskMessage{
 				"default": {t1},
 			},
@@ -275,9 +337,9 @@ func TestDequeueIgnoresPausedQueues(t *testing.T) {
 				"default":  {t1},
 				"critical": {t2},
 			},
-			args: []string{"default", "critical"},
-			want: nil,
-			err:  ErrNoProcessableTask,
+			args:    []string{"default", "critical"},
+			wantMsg: nil,
+			err:     ErrNoProcessableTask,
 			wantEnqueued: map[string][]*base.TaskMessage{
 				"default":  {t1},
 				"critical": {t2},
@@ -297,10 +359,10 @@ func TestDequeueIgnoresPausedQueues(t *testing.T) {
 			h.SeedEnqueuedQueue(t, r.client, msgs, queue)
 		}
 
-		got, err := r.Dequeue(tc.args...)
-		if !cmp.Equal(got, tc.want) || err != tc.err {
+		got, _, err := r.Dequeue(tc.args...)
+		if !cmp.Equal(got, tc.wantMsg) || err != tc.err {
 			t.Errorf("Dequeue(%v) = %v, %v; want %v, %v",
-				tc.args, got, err, tc.want, tc.err)
+				tc.args, got, err, tc.wantMsg, tc.err)
 			continue
 		}
 
@@ -320,40 +382,108 @@ func TestDequeueIgnoresPausedQueues(t *testing.T) {
 
 func TestDone(t *testing.T) {
 	r := setup(t)
-	t1 := h.NewTaskMessage("send_email", nil)
-	t2 := h.NewTaskMessage("export_csv", nil)
+	now := time.Now()
+	t1 := &base.TaskMessage{
+		ID:       xid.New(),
+		Type:     "send_email",
+		Payload:  nil,
+		Timeout:  1800,
+		Deadline: 0,
+	}
+	t1Deadline := now.Unix() + t1.Timeout
+	t2 := &base.TaskMessage{
+		ID:       xid.New(),
+		Type:     "export_csv",
+		Payload:  nil,
+		Timeout:  0,
+		Deadline: 1592485787,
+	}
+	t2Deadline := t2.Deadline
 	t3 := &base.TaskMessage{
 		ID:        xid.New(),
 		Type:      "reindex",
 		Payload:   nil,
+		Timeout:   1800,
+		Deadline:  0,
 		UniqueKey: "reindex:nil:default",
 		Queue:     "default",
 	}
+	t3Deadline := now.Unix() + t3.Deadline
 
 	tests := []struct {
 		inProgress     []*base.TaskMessage // initial state of the in-progress list
+		deadlines      []h.ZSetEntry       // initial state of deadlines set
 		target         *base.TaskMessage   // task to remove
 		wantInProgress []*base.TaskMessage // final state of the in-progress list
+		wantDeadlines  []h.ZSetEntry       // final state of the deadline set
 	}{
 		{
-			inProgress:     []*base.TaskMessage{t1, t2},
+			inProgress: []*base.TaskMessage{t1, t2},
+			deadlines: []h.ZSetEntry{
+				{
+					Msg:   t1,
+					Score: float64(t1Deadline),
+				},
+				{
+					Msg:   t2,
+					Score: float64(t2Deadline),
+				},
+			},
 			target:         t1,
 			wantInProgress: []*base.TaskMessage{t2},
+			wantDeadlines: []h.ZSetEntry{
+				{
+					Msg:   t2,
+					Score: float64(t2Deadline),
+				},
+			},
 		},
 		{
-			inProgress:     []*base.TaskMessage{t1},
+			inProgress: []*base.TaskMessage{t1},
+			deadlines: []h.ZSetEntry{
+				{
+					Msg:   t1,
+					Score: float64(t1Deadline),
+				},
+			},
 			target:         t1,
 			wantInProgress: []*base.TaskMessage{},
+			wantDeadlines:  []h.ZSetEntry{},
 		},
 		{
-			inProgress:     []*base.TaskMessage{t1, t2, t3},
+			inProgress: []*base.TaskMessage{t1, t2, t3},
+			deadlines: []h.ZSetEntry{
+				{
+					Msg:   t1,
+					Score: float64(t1Deadline),
+				},
+				{
+					Msg:   t2,
+					Score: float64(t2Deadline),
+				},
+				{
+					Msg:   t3,
+					Score: float64(t3Deadline),
+				},
+			},
 			target:         t3,
 			wantInProgress: []*base.TaskMessage{t1, t2},
+			wantDeadlines: []h.ZSetEntry{
+				{
+					Msg:   t1,
+					Score: float64(t1Deadline),
+				},
+				{
+					Msg:   t2,
+					Score: float64(t2Deadline),
+				},
+			},
 		},
 	}
 
 	for _, tc := range tests {
 		h.FlushDB(t, r.client) // clean up db before each test case
+		h.SeedDeadlines(t, r.client, tc.deadlines)
 		h.SeedInProgressQueue(t, r.client, tc.inProgress)
 		for _, msg := range tc.inProgress {
 			// Set uniqueness lock if unique key is present.
@@ -376,6 +506,11 @@ func TestDone(t *testing.T) {
 			t.Errorf("mismatch found in %q: (-want, +got):\n%s", base.InProgressQueue, diff)
 			continue
 		}
+		gotDeadlines := h.GetDeadlinesEntries(t, r.client)
+		if diff := cmp.Diff(tc.wantDeadlines, gotDeadlines, h.SortZSetEntryOpt); diff != "" {
+			t.Errorf("mismatch found in %q: (-want, +got):\n%s", base.KeyDeadlines, diff)
+			continue
+		}
 
 		processedKey := base.ProcessedKey(time.Now())
 		gotProcessed := r.client.Get(processedKey).Val()
@@ -396,38 +531,73 @@ func TestDone(t *testing.T) {
 
 func TestRequeue(t *testing.T) {
 	r := setup(t)
-	t1 := h.NewTaskMessage("send_email", nil)
-	t2 := h.NewTaskMessage("export_csv", nil)
-	t3 := h.NewTaskMessageWithQueue("send_email", nil, "critical")
+	now := time.Now()
+	t1 := &base.TaskMessage{
+		ID:      xid.New(),
+		Type:    "send_email",
+		Payload: nil,
+		Queue:   "default",
+		Timeout: 1800,
+	}
+	t2 := &base.TaskMessage{
+		ID:      xid.New(),
+		Type:    "export_csv",
+		Payload: nil,
+		Queue:   "default",
+		Timeout: 3000,
+	}
+	t3 := &base.TaskMessage{
+		ID:      xid.New(),
+		Type:    "send_email",
+		Payload: nil,
+		Queue:   "critical",
+		Timeout: 80,
+	}
+	t1Deadline := now.Unix() + t1.Timeout
+	t2Deadline := now.Unix() + t2.Timeout
+	t3Deadline := now.Unix() + t3.Timeout
 
 	tests := []struct {
 		enqueued       map[string][]*base.TaskMessage // initial state of queues
 		inProgress     []*base.TaskMessage            // initial state of the in-progress list
+		deadlines      []h.ZSetEntry                  // initial state of the deadlines set
 		target         *base.TaskMessage              // task to requeue
 		wantEnqueued   map[string][]*base.TaskMessage // final state of queues
 		wantInProgress []*base.TaskMessage            // final state of the in-progress list
+		wantDeadlines  []h.ZSetEntry                  // final state of the deadlines set
 	}{
 		{
 			enqueued: map[string][]*base.TaskMessage{
 				base.DefaultQueueName: {},
 			},
 			inProgress: []*base.TaskMessage{t1, t2},
-			target:     t1,
+			deadlines: []h.ZSetEntry{
+				{Msg: t1, Score: float64(t1Deadline)},
+				{Msg: t2, Score: float64(t2Deadline)},
+			},
+			target: t1,
 			wantEnqueued: map[string][]*base.TaskMessage{
 				base.DefaultQueueName: {t1},
 			},
 			wantInProgress: []*base.TaskMessage{t2},
+			wantDeadlines: []h.ZSetEntry{
+				{Msg: t2, Score: float64(t2Deadline)},
+			},
 		},
 		{
 			enqueued: map[string][]*base.TaskMessage{
 				base.DefaultQueueName: {t1},
 			},
 			inProgress: []*base.TaskMessage{t2},
-			target:     t2,
+			deadlines: []h.ZSetEntry{
+				{Msg: t2, Score: float64(t2Deadline)},
+			},
+			target: t2,
 			wantEnqueued: map[string][]*base.TaskMessage{
 				base.DefaultQueueName: {t1, t2},
 			},
 			wantInProgress: []*base.TaskMessage{},
+			wantDeadlines:  []h.ZSetEntry{},
 		},
 		{
 			enqueued: map[string][]*base.TaskMessage{
@@ -435,12 +605,19 @@ func TestRequeue(t *testing.T) {
 				"critical":            {},
 			},
 			inProgress: []*base.TaskMessage{t2, t3},
-			target:     t3,
+			deadlines: []h.ZSetEntry{
+				{Msg: t2, Score: float64(t2Deadline)},
+				{Msg: t3, Score: float64(t3Deadline)},
+			},
+			target: t3,
 			wantEnqueued: map[string][]*base.TaskMessage{
 				base.DefaultQueueName: {t1},
 				"critical":            {t3},
 			},
 			wantInProgress: []*base.TaskMessage{t2},
+			wantDeadlines: []h.ZSetEntry{
+				{Msg: t2, Score: float64(t2Deadline)},
+			},
 		},
 	}
 
@@ -450,6 +627,7 @@ func TestRequeue(t *testing.T) {
 			h.SeedEnqueuedQueue(t, r.client, msgs, qname)
 		}
 		h.SeedInProgressQueue(t, r.client, tc.inProgress)
+		h.SeedDeadlines(t, r.client, tc.deadlines)
 
 		err := r.Requeue(tc.target)
 		if err != nil {
@@ -468,6 +646,10 @@ func TestRequeue(t *testing.T) {
 		if diff := cmp.Diff(tc.wantInProgress, gotInProgress, h.SortMsgOpt); diff != "" {
 			t.Errorf("mismatch found in %q: (-want, +got):\n%s", base.InProgressQueue, diff)
 		}
+		gotDeadlines := h.GetDeadlinesEntries(t, r.client)
+		if diff := cmp.Diff(tc.wantDeadlines, gotDeadlines, h.SortZSetEntryOpt); diff != "" {
+			t.Errorf("mismatch found in %q: (-want, +got):\n%s", base.KeyDeadlines, diff)
+		}
 	}
 }
 
@@ -557,33 +739,47 @@ func TestScheduleUnique(t *testing.T) {
 
 func TestRetry(t *testing.T) {
 	r := setup(t)
-	t1 := h.NewTaskMessage("send_email", map[string]interface{}{"subject": "Hola!"})
-	t2 := h.NewTaskMessage("gen_thumbnail", map[string]interface{}{"path": "some/path/to/image.jpg"})
-	t3 := h.NewTaskMessage("reindex", nil)
-	t1.Retried = 10
-	errMsg := "SMTP server is not responding"
-	t1AfterRetry := &base.TaskMessage{
-		ID:       t1.ID,
-		Type:     t1.Type,
-		Payload:  t1.Payload,
-		Queue:    t1.Queue,
-		Retry:    t1.Retry,
-		Retried:  t1.Retried + 1,
-		ErrorMsg: errMsg,
-	}
 	now := time.Now()
+	t1 := &base.TaskMessage{
+		ID:      xid.New(),
+		Type:    "send_email",
+		Payload: map[string]interface{}{"subject": "Hola!"},
+		Retried: 10,
+		Timeout: 1800,
+	}
+	t1Deadline := now.Unix() + t1.Timeout
+	t2 := &base.TaskMessage{
+		ID:      xid.New(),
+		Type:    "gen_thumbnail",
+		Payload: map[string]interface{}{"path": "some/path/to/image.jpg"},
+		Timeout: 3000,
+	}
+	t2Deadline := now.Unix() + t2.Timeout
+	t3 := &base.TaskMessage{
+		ID:      xid.New(),
+		Type:    "reindex",
+		Payload: nil,
+		Timeout: 60,
+	}
+	errMsg := "SMTP server is not responding"
 
 	tests := []struct {
 		inProgress     []*base.TaskMessage
+		deadlines      []h.ZSetEntry
 		retry          []h.ZSetEntry
 		msg            *base.TaskMessage
 		processAt      time.Time
 		errMsg         string
 		wantInProgress []*base.TaskMessage
+		wantDeadlines  []h.ZSetEntry
 		wantRetry      []h.ZSetEntry
 	}{
 		{
 			inProgress: []*base.TaskMessage{t1, t2},
+			deadlines: []h.ZSetEntry{
+				{Msg: t1, Score: float64(t1Deadline)},
+				{Msg: t2, Score: float64(t2Deadline)},
+			},
 			retry: []h.ZSetEntry{
 				{
 					Msg:   t3,
@@ -594,9 +790,12 @@ func TestRetry(t *testing.T) {
 			processAt:      now.Add(5 * time.Minute),
 			errMsg:         errMsg,
 			wantInProgress: []*base.TaskMessage{t2},
+			wantDeadlines: []h.ZSetEntry{
+				{Msg: t2, Score: float64(t2Deadline)},
+			},
 			wantRetry: []h.ZSetEntry{
 				{
-					Msg:   t1AfterRetry,
+					Msg:   h.TaskMessageAfterRetry(*t1, errMsg),
 					Score: float64(now.Add(5 * time.Minute).Unix()),
 				},
 				{
@@ -610,6 +809,7 @@ func TestRetry(t *testing.T) {
 	for _, tc := range tests {
 		h.FlushDB(t, r.client)
 		h.SeedInProgressQueue(t, r.client, tc.inProgress)
+		h.SeedDeadlines(t, r.client, tc.deadlines)
 		h.SeedRetryQueue(t, r.client, tc.retry)
 
 		err := r.Retry(tc.msg, tc.processAt, tc.errMsg)
@@ -622,7 +822,10 @@ func TestRetry(t *testing.T) {
 		if diff := cmp.Diff(tc.wantInProgress, gotInProgress, h.SortMsgOpt); diff != "" {
 			t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.InProgressQueue, diff)
 		}
-
+		gotDeadlines := h.GetDeadlinesEntries(t, r.client)
+		if diff := cmp.Diff(tc.wantDeadlines, gotDeadlines, h.SortZSetEntryOpt); diff != "" {
+			t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.KeyDeadlines, diff)
+		}
 		gotRetry := h.GetRetryEntries(t, r.client)
 		if diff := cmp.Diff(tc.wantRetry, gotRetry, h.SortZSetEntryOpt); diff != "" {
 			t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.RetryQueue, diff)
@@ -652,31 +855,55 @@ func TestRetry(t *testing.T) {
 
 func TestKill(t *testing.T) {
 	r := setup(t)
-	t1 := h.NewTaskMessage("send_email", nil)
-	t2 := h.NewTaskMessage("reindex", nil)
-	t3 := h.NewTaskMessage("generate_csv", nil)
-	errMsg := "SMTP server not responding"
-	t1AfterKill := &base.TaskMessage{
-		ID:       t1.ID,
-		Type:     t1.Type,
-		Payload:  t1.Payload,
-		Queue:    t1.Queue,
-		Retry:    t1.Retry,
-		Retried:  t1.Retried,
-		ErrorMsg: errMsg,
-	}
 	now := time.Now()
+	t1 := &base.TaskMessage{
+		ID:      xid.New(),
+		Type:    "send_email",
+		Payload: nil,
+		Queue:   "default",
+		Retry:   25,
+		Retried: 0,
+		Timeout: 1800,
+	}
+	t1Deadline := now.Unix() + t1.Timeout
+	t2 := &base.TaskMessage{
+		ID:      xid.New(),
+		Type:    "reindex",
+		Payload: nil,
+		Queue:   "default",
+		Retry:   25,
+		Retried: 0,
+		Timeout: 3000,
+	}
+	t2Deadline := now.Unix() + t2.Timeout
+	t3 := &base.TaskMessage{
+		ID:      xid.New(),
+		Type:    "generate_csv",
+		Payload: nil,
+		Queue:   "default",
+		Retry:   25,
+		Retried: 0,
+		Timeout: 60,
+	}
+	t3Deadline := now.Unix() + t3.Timeout
+	errMsg := "SMTP server not responding"
 
 	// TODO(hibiken): add test cases for trimming
 	tests := []struct {
 		inProgress     []*base.TaskMessage
+		deadlines      []h.ZSetEntry
 		dead           []h.ZSetEntry
 		target         *base.TaskMessage // task to kill
 		wantInProgress []*base.TaskMessage
+		wantDeadlines  []h.ZSetEntry
 		wantDead       []h.ZSetEntry
 	}{
 		{
 			inProgress: []*base.TaskMessage{t1, t2},
+			deadlines: []h.ZSetEntry{
+				{Msg: t1, Score: float64(t1Deadline)},
+				{Msg: t2, Score: float64(t2Deadline)},
+			},
 			dead: []h.ZSetEntry{
 				{
 					Msg:   t3,
@@ -685,9 +912,12 @@ func TestKill(t *testing.T) {
 			},
 			target:         t1,
 			wantInProgress: []*base.TaskMessage{t2},
+			wantDeadlines: []h.ZSetEntry{
+				{Msg: t2, Score: float64(t2Deadline)},
+			},
 			wantDead: []h.ZSetEntry{
 				{
-					Msg:   t1AfterKill,
+					Msg:   h.TaskMessageWithError(*t1, errMsg),
 					Score: float64(now.Unix()),
 				},
 				{
@@ -697,13 +927,22 @@ func TestKill(t *testing.T) {
 			},
 		},
 		{
-			inProgress:     []*base.TaskMessage{t1, t2, t3},
+			inProgress: []*base.TaskMessage{t1, t2, t3},
+			deadlines: []h.ZSetEntry{
+				{Msg: t1, Score: float64(t1Deadline)},
+				{Msg: t2, Score: float64(t2Deadline)},
+				{Msg: t3, Score: float64(t3Deadline)},
+			},
 			dead:           []h.ZSetEntry{},
 			target:         t1,
 			wantInProgress: []*base.TaskMessage{t2, t3},
+			wantDeadlines: []h.ZSetEntry{
+				{Msg: t2, Score: float64(t2Deadline)},
+				{Msg: t3, Score: float64(t3Deadline)},
+			},
 			wantDead: []h.ZSetEntry{
 				{
-					Msg:   t1AfterKill,
+					Msg:   h.TaskMessageWithError(*t1, errMsg),
 					Score: float64(now.Unix()),
 				},
 			},
@@ -713,6 +952,7 @@ func TestKill(t *testing.T) {
 	for _, tc := range tests {
 		h.FlushDB(t, r.client) // clean up db before each test case
 		h.SeedInProgressQueue(t, r.client, tc.inProgress)
+		h.SeedDeadlines(t, r.client, tc.deadlines)
 		h.SeedDeadQueue(t, r.client, tc.dead)
 
 		err := r.Kill(tc.target, errMsg)
@@ -725,7 +965,10 @@ func TestKill(t *testing.T) {
 		if diff := cmp.Diff(tc.wantInProgress, gotInProgress, h.SortMsgOpt); diff != "" {
 			t.Errorf("mismatch found in %q: (-want, +got)\n%s", base.InProgressQueue, diff)
 		}
-
+		gotDeadlines := h.GetDeadlinesEntries(t, r.client)
+		if diff := cmp.Diff(tc.wantDeadlines, gotDeadlines, h.SortZSetEntryOpt); diff != "" {
+			t.Errorf("mismatch found in %q after calling (*RDB).Kill: (-want, +got):\n%s", base.KeyDeadlines, diff)
+		}
 		gotDead := h.GetDeadEntries(t, r.client)
 		if diff := cmp.Diff(tc.wantDead, gotDead, h.SortZSetEntryOpt); diff != "" {
 			t.Errorf("mismatch found in %q after calling (*RDB).Kill: (-want, +got):\n%s", base.DeadQueue, diff)
@@ -856,6 +1099,77 @@ func TestCheckAndEnqueue(t *testing.T) {
 	}
 }
 
+func TestListDeadlineExceeded(t *testing.T) {
+	t1 := h.NewTaskMessage("task1", nil)
+	t2 := h.NewTaskMessage("task2", nil)
+	t3 := h.NewTaskMessageWithQueue("task3", nil, "critical")
+
+	now := time.Now()
+	oneHourFromNow := now.Add(1 * time.Hour)
+	fiveMinutesFromNow := now.Add(5 * time.Minute)
+	fiveMinutesAgo := now.Add(-5 * time.Minute)
+	oneHourAgo := now.Add(-1 * time.Hour)
+
+	tests := []struct {
+		desc      string
+		deadlines []h.ZSetEntry
+		t         time.Time
+		want      []*base.TaskMessage
+	}{
+		{
+			desc: "with one task in-progress",
+			deadlines: []h.ZSetEntry{
+				{Msg: t1, Score: float64(fiveMinutesAgo.Unix())},
+			},
+			t:    time.Now(),
+			want: []*base.TaskMessage{t1},
+		},
+		{
+			desc: "with multiple tasks in-progress, and one expired",
+			deadlines: []h.ZSetEntry{
+				{Msg: t1, Score: float64(oneHourAgo.Unix())},
+				{Msg: t2, Score: float64(fiveMinutesFromNow.Unix())},
+				{Msg: t3, Score: float64(oneHourFromNow.Unix())},
+			},
+			t:    time.Now(),
+			want: []*base.TaskMessage{t1},
+		},
+		{
+			desc: "with multiple expired tasks in-progress",
+			deadlines: []h.ZSetEntry{
+				{Msg: t1, Score: float64(oneHourAgo.Unix())},
+				{Msg: t2, Score: float64(fiveMinutesAgo.Unix())},
+				{Msg: t3, Score: float64(oneHourFromNow.Unix())},
+			},
+			t:    time.Now(),
+			want: []*base.TaskMessage{t1, t2},
+		},
+		{
+			desc:      "with empty in-progress queue",
+			deadlines: []h.ZSetEntry{},
+			t:         time.Now(),
+			want:      []*base.TaskMessage{},
+		},
+	}
+
+	r := setup(t)
+	for _, tc := range tests {
+		h.FlushDB(t, r.client)
+		h.SeedDeadlines(t, r.client, tc.deadlines)
+
+		got, err := r.ListDeadlineExceeded(tc.t)
+		if err != nil {
+			t.Errorf("%s; ListDeadlineExceeded(%v) returned error: %v", tc.desc, tc.t, err)
+			continue
+		}
+
+		if diff := cmp.Diff(tc.want, got, h.SortMsgOpt); diff != "" {
+			t.Errorf("%s; ListDeadlineExceeded(%v) returned %v, want %v;(-want,+got)\n%s",
+				tc.desc, tc.t, got, tc.want, diff)
+		}
+	}
+}
+
 func TestWriteServerState(t *testing.T) {
 	r := setup(t)
 
diff --git a/internal/testbroker/testbroker.go b/internal/testbroker/testbroker.go
index 08e407f..85d388b 100644
--- a/internal/testbroker/testbroker.go
+++ b/internal/testbroker/testbroker.go
@@ -26,6 +26,9 @@ type TestBroker struct {
 	real base.Broker
 }
 
+// Make sure TestBroker implements Broker interface at compile time.
+var _ base.Broker = (*TestBroker)(nil)
+
 func NewTestBroker(b base.Broker) *TestBroker {
 	return &TestBroker{real: b}
 }
@@ -60,11 +63,11 @@ func (tb *TestBroker) EnqueueUnique(msg *base.TaskMessage, ttl time.Duration) er
 	return tb.real.EnqueueUnique(msg, ttl)
 }
 
-func (tb *TestBroker) Dequeue(qnames ...string) (*base.TaskMessage, error) {
+func (tb *TestBroker) Dequeue(qnames ...string) (*base.TaskMessage, time.Time, error) {
 	tb.mu.Lock()
 	defer tb.mu.Unlock()
 	if tb.sleeping {
-		return nil, errRedisDown
+		return nil, time.Time{}, errRedisDown
 	}
 	return tb.real.Dequeue(qnames...)
 }
@@ -132,6 +135,15 @@ func (tb *TestBroker) CheckAndEnqueue() error {
 	return tb.real.CheckAndEnqueue()
 }
 
+func (tb *TestBroker) ListDeadlineExceeded(deadline time.Time) ([]*base.TaskMessage, error) {
+	tb.mu.Lock()
+	defer tb.mu.Unlock()
+	if tb.sleeping {
+		return nil, errRedisDown
+	}
+	return tb.real.ListDeadlineExceeded(deadline)
+}
+
 func (tb *TestBroker) WriteServerState(info *base.ServerInfo, workers []*base.WorkerInfo, ttl time.Duration) error {
 	tb.mu.Lock()
 	defer tb.mu.Unlock()
diff --git a/processor_test.go b/processor_test.go
index 7a7c8a0..0b7183b 100644
--- a/processor_test.go
+++ b/processor_test.go
@@ -215,19 +215,6 @@ func TestProcessorRetry(t *testing.T) {
 	m4 := h.NewTaskMessage("sync", nil)
 
 	errMsg := "something went wrong"
-	// r* is m* after retry
-	r1 := *m1
-	r1.ErrorMsg = errMsg
-	r2 := *m2
-	r2.ErrorMsg = errMsg
-	r2.Retried = m2.Retried + 1
-	r3 := *m3
-	r3.ErrorMsg = errMsg
-	r3.Retried = m3.Retried + 1
-	r4 := *m4
-	r4.ErrorMsg = errMsg
-	r4.Retried = m4.Retried + 1
-
 	now := time.Now()
 
 	tests := []struct {
@@ -249,11 +236,11 @@ func TestProcessorRetry(t *testing.T) {
 			}),
 			wait: 2 * time.Second,
 			wantRetry: []h.ZSetEntry{
-				{Msg: &r2, Score: float64(now.Add(time.Minute).Unix())},
-				{Msg: &r3, Score: float64(now.Add(time.Minute).Unix())},
-				{Msg: &r4, Score: float64(now.Add(time.Minute).Unix())},
+				{Msg: h.TaskMessageAfterRetry(*m2, errMsg), Score: float64(now.Add(time.Minute).Unix())},
+				{Msg: h.TaskMessageAfterRetry(*m3, errMsg), Score: float64(now.Add(time.Minute).Unix())},
+				{Msg: h.TaskMessageAfterRetry(*m4, errMsg), Score: float64(now.Add(time.Minute).Unix())},
 			},
-			wantDead:     []*base.TaskMessage{&r1},
+			wantDead:     []*base.TaskMessage{h.TaskMessageWithError(*m1, errMsg)},
 			wantErrCount: 4,
 		},
 	}
diff --git a/recoverer_test.go b/recoverer_test.go
new file mode 100644
index 0000000..5d33f9a
--- /dev/null
+++ b/recoverer_test.go
@@ -0,0 +1,162 @@
+// Copyright 2020 Kentaro Hibino. All rights reserved.
+// Use of this source code is governed by a MIT license
+// that can be found in the LICENSE file.
+
+package asynq
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+	h "github.com/hibiken/asynq/internal/asynqtest"
+	"github.com/hibiken/asynq/internal/base"
+	"github.com/hibiken/asynq/internal/rdb"
+)
+
+func TestRecoverer(t *testing.T) {
+	r := setup(t)
+	rdbClient := rdb.NewRDB(r)
+
+	t1 := h.NewTaskMessage("task1", nil)
+	t2 := h.NewTaskMessage("task2", nil)
+	t3 := h.NewTaskMessageWithQueue("task3", nil, "critical")
+	t4 := h.NewTaskMessage("task4", nil)
+	t4.Retried = t4.Retry // t4 has reached its max retry count
+
+	now := time.Now()
+	oneHourFromNow := now.Add(1 * time.Hour)
+	fiveMinutesFromNow := now.Add(5 * time.Minute)
+	fiveMinutesAgo := now.Add(-5 * time.Minute)
+	oneHourAgo := now.Add(-1 * time.Hour)
+
+	tests := []struct {
+		desc           string
+		inProgress     []*base.TaskMessage
+		deadlines      []h.ZSetEntry
+		retry          []h.ZSetEntry
+		dead           []h.ZSetEntry
+		wantInProgress []*base.TaskMessage
+		wantDeadlines  []h.ZSetEntry
+		wantRetry      []*base.TaskMessage
+		wantDead       []*base.TaskMessage
+	}{
+		{
+			desc:       "with one task in-progress",
+			inProgress: []*base.TaskMessage{t1},
+			deadlines: []h.ZSetEntry{
+				{Msg: t1, Score: float64(fiveMinutesAgo.Unix())},
+			},
+			retry:          []h.ZSetEntry{},
+			dead:           []h.ZSetEntry{},
+			wantInProgress: []*base.TaskMessage{},
+			wantDeadlines:  []h.ZSetEntry{},
+			wantRetry: []*base.TaskMessage{
+				h.TaskMessageAfterRetry(*t1, "deadline exceeded"),
+			},
+			wantDead: []*base.TaskMessage{},
+		},
+		{
+			desc:       "with a task with max-retry reached",
+			inProgress: []*base.TaskMessage{t4},
+			deadlines: []h.ZSetEntry{
+				{Msg: t4, Score: float64(fiveMinutesAgo.Unix())},
+			},
+			retry:          []h.ZSetEntry{},
+			dead:           []h.ZSetEntry{},
+			wantInProgress: []*base.TaskMessage{},
+			wantDeadlines:  []h.ZSetEntry{},
+			wantRetry:      []*base.TaskMessage{},
+			wantDead:       []*base.TaskMessage{h.TaskMessageWithError(*t4, "deadline exceeded")},
+		},
+		{
+			desc:       "with multiple tasks in-progress, and one expired",
+			inProgress: []*base.TaskMessage{t1, t2, t3},
+			deadlines: []h.ZSetEntry{
+				{Msg: t1, Score: float64(oneHourAgo.Unix())},
+				{Msg: t2, Score: float64(fiveMinutesFromNow.Unix())},
+				{Msg: t3, Score: float64(oneHourFromNow.Unix())},
+			},
+			retry:          []h.ZSetEntry{},
+			dead:           []h.ZSetEntry{},
+			wantInProgress: []*base.TaskMessage{t2, t3},
+			wantDeadlines: []h.ZSetEntry{
+				{Msg: t2, Score: float64(fiveMinutesFromNow.Unix())},
+				{Msg: t3, Score: float64(oneHourFromNow.Unix())},
+			},
+			wantRetry: []*base.TaskMessage{
+				h.TaskMessageAfterRetry(*t1, "deadline exceeded"),
+			},
+			wantDead: []*base.TaskMessage{},
+		},
+		{
+			desc:       "with multiple expired tasks in-progress",
+			inProgress: []*base.TaskMessage{t1, t2, t3},
+			deadlines: []h.ZSetEntry{
+				{Msg: t1, Score: float64(oneHourAgo.Unix())},
+				{Msg: t2, Score: float64(fiveMinutesAgo.Unix())},
+				{Msg: t3, Score: float64(oneHourFromNow.Unix())},
+			},
+			retry:          []h.ZSetEntry{},
+			dead:           []h.ZSetEntry{},
+			wantInProgress: []*base.TaskMessage{t3},
+			wantDeadlines: []h.ZSetEntry{
+				{Msg: t3, Score: float64(oneHourFromNow.Unix())},
+			},
+			wantRetry: []*base.TaskMessage{
+				h.TaskMessageAfterRetry(*t1, "deadline exceeded"),
+				h.TaskMessageAfterRetry(*t2, "deadline exceeded"),
+			},
+			wantDead: []*base.TaskMessage{},
+		},
+		{
+			desc:           "with empty in-progress queue",
+			inProgress:     []*base.TaskMessage{},
+			deadlines:      []h.ZSetEntry{},
+			retry:          []h.ZSetEntry{},
+			dead:           []h.ZSetEntry{},
+			wantInProgress: []*base.TaskMessage{},
+			wantDeadlines:  []h.ZSetEntry{},
+			wantRetry:      []*base.TaskMessage{},
+			wantDead:       []*base.TaskMessage{},
+		},
+	}
+
+	for _, tc := range tests {
+		h.FlushDB(t, r)
+		h.SeedInProgressQueue(t, r, tc.inProgress)
+		h.SeedDeadlines(t, r, tc.deadlines)
+		h.SeedRetryQueue(t, r, tc.retry)
+		h.SeedDeadQueue(t, r, tc.dead)
+
+		recoverer := newRecoverer(recovererParams{
+			logger:         testLogger,
+			broker:         rdbClient,
+			interval:       1 * time.Second,
+			retryDelayFunc: func(n int, err error, task *Task) time.Duration { return 30 * time.Second },
+		})
+
+		var wg sync.WaitGroup
+		recoverer.start(&wg)
+		time.Sleep(2 * time.Second)
+		recoverer.terminate()
+
+		gotInProgress := h.GetInProgressMessages(t, r)
+		if diff := cmp.Diff(tc.wantInProgress, gotInProgress, h.SortMsgOpt); diff != "" {
+			t.Errorf("%s; mismatch found in %q; (-want,+got)\n%s", tc.desc, base.InProgressQueue, diff)
+		}
+		gotDeadlines := h.GetDeadlinesEntries(t, r)
+		if diff := cmp.Diff(tc.wantDeadlines, gotDeadlines, h.SortZSetEntryOpt); diff != "" {
+			t.Errorf("%s; mismatch found in %q; (-want,+got)\n%s", tc.desc, base.KeyDeadlines, diff)
+		}
+		gotRetry := h.GetRetryMessages(t, r)
+		if diff := cmp.Diff(tc.wantRetry, gotRetry, h.SortMsgOpt); diff != "" {
+			t.Errorf("%s; mismatch found in %q: (-want, +got)\n%s", tc.desc, base.RetryQueue, diff)
+		}
+		gotDead := h.GetDeadMessages(t, r)
+		if diff := cmp.Diff(tc.wantDead, gotDead, h.SortMsgOpt); diff != "" {
+			t.Errorf("%s; mismatch found in %q: (-want, +got)\n%s", tc.desc, base.DeadQueue, diff)
+		}
+	}
+}
diff --git a/syncer_test.go b/syncer_test.go
index 85f68cb..1be2d18 100644
--- a/syncer_test.go
+++ b/syncer_test.go
@@ -42,6 +42,7 @@ func TestSyncer(t *testing.T) {
 			fn: func() error {
 				return rdbClient.Done(m)
 			},
+			deadline: time.Now().Add(5 * time.Minute),
 		}
 	}
 
@@ -85,8 +86,9 @@ func TestSyncerRetry(t *testing.T) {
 	}
 
 	syncRequestCh <- &syncRequest{
-		fn:     requestFunc,
-		errMsg: "error",
+		fn:       requestFunc,
+		errMsg:   "error",
+		deadline: time.Now().Add(5 * time.Minute),
 	}
 
 	// allow syncer to retry
@@ -98,3 +100,41 @@ func TestSyncerRetry(t *testing.T) {
 	}
 	mu.Unlock()
 }
+
+func TestSyncerDropsStaleRequests(t *testing.T) {
+	const interval = time.Second
+	syncRequestCh := make(chan *syncRequest)
+	syncer := newSyncer(syncerParams{
+		logger:     testLogger,
+		requestsCh: syncRequestCh,
+		interval:   interval,
+	})
+	var wg sync.WaitGroup
+	syncer.start(&wg)
+
+	var (
+		mu sync.Mutex
+		n  int // number of times request has been processed
+	)
+
+	for i := 0; i < 10; i++ {
+		syncRequestCh <- &syncRequest{
+			fn: func() error {
+				mu.Lock()
+				n++
+				mu.Unlock()
+				return nil
+			},
+			deadline: time.Now().Add(time.Duration(-i) * time.Second), // already exceeded deadline
+		}
+	}
+
+	time.Sleep(2 * interval) // ensure that syncer runs at least once
+	syncer.terminate()
+
+	mu.Lock()
+	if n != 0 {
+		t.Errorf("requests has been processed %d times, want 0", n)
+	}
+	mu.Unlock()
+}