Repository: hibiken/asynq
Pull request #181 (2020-07-06 12:29:40)

v0.10.0
#### Added Features

- Automatic recovery of tasks in the event of a worker crash
- Automatic retry of tasks which exceeded its timeout/deadline


#### Version 0.10.0 includes the following API changes:

`(*Client).Enqueue`, `(*Client).EnqueueIn`, and `(*Client).EnqueueAt` has changed to return a `*Result` and error. A Result struct contains metadata about task that was enqueued (e.g. ID, Queue, etc).
`ErrorHandler` signature has changed to `func(context.Context, *Task, error)`. 
 
#### Version 0.10.0 includes the following semantics changes:

All tasks now require timeout or deadline. By default, timeout is set to 1800 seconds(30 mins) if neither of them are specified.
Tasks that exceed its deadline are automatically retried. In the previous versions, User provided Handler needed to explicitly return an error when ctx.Done channel is closed. In the new version, this is taken care of by the library. In order to avoid processing tasks when its deadline is exceeded, Handler should always check ctx.Done channel and stop processing when the channel is closed.


#### Other important changes:

Encoding schema for messages has changed. Install the latest CLI and run migrate command if you have tasks enqueued with the previous version of asynq.

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

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

```diff
@@            Coverage Diff             @@
##           master     #181      +/-   ##
==========================================
- Coverage   84.04%   83.40%   -0.65%     
==========================================
  Files          16       17       +1     
  Lines        1598     1699     +101     
==========================================
+ Hits         1343     1417      +74     
- Misses        172      189      +17     
- Partials       83       93      +10     
```


| [Impacted Files](https://codecov.io/gh/hibiken/asynq/pull/181?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [internal/base/base.go](https://codecov.io/gh/hibiken/asynq/pull/181/diff?src=pr&el=tree#diff-aW50ZXJuYWwvYmFzZS9iYXNlLmdv) | `90.90% <ø> (-1.03%)` | :arrow_down: |
| [internal/rdb/rdb.go](https://codecov.io/gh/hibiken/asynq/pull/181/diff?src=pr&el=tree#diff-aW50ZXJuYWwvcmRiL3JkYi5nbw==) | `70.83% <66.66%> (-3.56%)` | :arrow_down: |
| [processor.go](https://codecov.io/gh/hibiken/asynq/pull/181/diff?src=pr&el=tree#diff-cHJvY2Vzc29yLmdv) | `88.46% <75.47%> (-2.02%)` | :arrow_down: |
| [recoverer.go](https://codecov.io/gh/hibiken/asynq/pull/181/diff?src=pr&el=tree#diff-cmVjb3ZlcmVyLmdv) | `82.92% <82.92%> (ø)` | |
| [client.go](https://codecov.io/gh/hibiken/asynq/pull/181/diff?src=pr&el=tree#diff-Y2xpZW50Lmdv) | `95.79% <92.85%> (-1.27%)` | :arrow_down: |
| [context.go](https://codecov.io/gh/hibiken/asynq/pull/181/diff?src=pr&el=tree#diff-Y29udGV4dC5nbw==) | `100.00% <100.00%> (ø)` | |
| [heartbeat.go](https://codecov.io/gh/hibiken/asynq/pull/181/diff?src=pr&el=tree#diff-aGVhcnRiZWF0Lmdv) | `97.01% <100.00%> (ø)` | |
| [internal/rdb/inspect.go](https://codecov.io/gh/hibiken/asynq/pull/181/diff?src=pr&el=tree#diff-aW50ZXJuYWwvcmRiL2luc3BlY3QuZ28=) | `76.53% <100.00%> (ø)` | |
| [server.go](https://codecov.io/gh/hibiken/asynq/pull/181/diff?src=pr&el=tree#diff-c2VydmVyLmdv) | `85.22% <100.00%> (+0.79%)` | :arrow_up: |
| [syncer.go](https://codecov.io/gh/hibiken/asynq/pull/181/diff?src=pr&el=tree#diff-c3luY2VyLmdv) | `100.00% <100.00%> (+6.66%)` | :arrow_up: |
| ... and [2 more](https://codecov.io/gh/hibiken/asynq/pull/181/diff?src=pr&el=tree-more) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/hibiken/asynq/pull/181?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/181?src=pr&el=footer). Last update [4e5f596...15f3afc](https://codecov.io/gh/hibiken/asynq/pull/181?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..07459ff 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,16 @@ 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 latest CLI and run `migrate` command if 
+  you have tasks enqueued with the previous version of asynq.
+- API of `(*Client).Enqueue`, `(*Client).EnqueueIn`, and `(*Client).EnqueueAt` has changed to return a `*Result`.
+- API of `ErrorHandler` has changed. It now takes context as the first argument and removed `retried`, `maxRetry` from the argument list.
+  Use `GetRetryCount` and/or `GetMaxRetry` to get the count values.
+
 ## [0.9.4] - 2020-06-13
 
 ### Fixed
diff --git a/README.md b/README.md
index 44899ee..6bdf822 100644
--- a/README.md
+++ b/README.md
@@ -34,6 +34,7 @@ A system can consist of multiple worker servers and brokers, giving way to high
 - Scheduling of tasks
 - Durability since tasks are written to Redis
 - [Retries](https://github.com/hibiken/asynq/wiki/Task-Retry) of failed tasks
+- Automatic recovery of tasks in the event of a worker crash
 - [Weighted priority queues](https://github.com/hibiken/asynq/wiki/Priority-Queues#weighted-priority-queues)
 - [Strict priority queues](https://github.com/hibiken/asynq/wiki/Priority-Queues#strict-priority-queues)
 - Low latency to add a task since writes are fast in Redis
@@ -157,10 +158,11 @@ func main() {
     // ------------------------------------------------------
 
     t := tasks.NewEmailDeliveryTask(42, "some:template:id")
-    err := c.Enqueue(t)
+    res, err := c.Enqueue(t)
     if err != nil {
         log.Fatal("could not enqueue task: %v", err)
     }
+    fmt.Printf("Enqueued Result: %+v\n", res)
 
 
     // ------------------------------------------------------------
@@ -169,10 +171,11 @@ func main() {
     // ------------------------------------------------------------
 
     t = tasks.NewEmailDeliveryTask(42, "other:template:id")
-    err = c.EnqueueIn(24*time.Hour, t)
+    res, err = c.EnqueueIn(24*time.Hour, t)
     if err != nil {
         log.Fatal("could not schedule task: %v", err)
     }
+    fmt.Printf("Enqueued Result: %+v\n", res)
 
 
     // ----------------------------------------------------------------------------
@@ -183,10 +186,11 @@ func main() {
     c.SetDefaultOptions(tasks.ImageProcessing, asynq.MaxRetry(10), asynq.Timeout(time.Minute))
 
     t = tasks.NewImageProcessingTask("some/blobstore/url", "other/blobstore/url")
-    err = c.Enqueue(t)
+    res, err = c.Enqueue(t)
     if err != nil {
         log.Fatal("could not enqueue task: %v", err)
     }
+    fmt.Printf("Enqueued Result: %+v\n", res)
 
     // ---------------------------------------------------------------------------
     // Example 4: Pass options to tune task processing behavior at enqueue time.
@@ -194,10 +198,11 @@ func main() {
     // ---------------------------------------------------------------------------
 
     t = tasks.NewImageProcessingTask("some/blobstore/url", "other/blobstore/url")
-    err = c.Enqueue(t, asynq.Queue("critical"), asynq.Timeout(30*time.Second))
+    res, err = c.Enqueue(t, asynq.Queue("critical"), asynq.Timeout(30*time.Second))
     if err != nil {
         log.Fatal("could not enqueue task: %v", err)
     }
+    fmt.Printf("Enqueued Result: %+v\n", res)
 }
 ```
 
diff --git a/client.go b/client.go
index af0523a..6ac7a72 100644
--- a/client.go
+++ b/client.go
@@ -12,9 +12,9 @@ import (
 	"sync"
 	"time"
 
+	"github.com/google/uuid"
 	"github.com/hibiken/asynq/internal/base"
 	"github.com/hibiken/asynq/internal/rdb"
-	"github.com/rs/xid"
 )
 
 // A Client is responsible for scheduling tasks.
@@ -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.
@@ -179,13 +200,43 @@ func (c *Client) SetDefaultOptions(taskType string, opts ...Option) {
 	c.opts[taskType] = opts
 }
 
+// A Result holds enqueued task's metadata.
+type Result struct {
+	// ID is a unique identifier for the task.
+	ID string
+
+	// Retry is the maximum number of retry for the task.
+	Retry int
+
+	// Queue is a name of the queue the task is enqueued to.
+	Queue string
+
+	// Timeout is the timeout value for the task.
+	// Counting for timeout starts when a worker starts processing the task.
+	// If task processing doesn't complete within the timeout, the task will be retried.
+	// The value zero means no timeout.
+	//
+	// If deadline is set, min(now+timeout, deadline) is used, where the now is the time when
+	// a worker starts processing the task.
+	Timeout time.Duration
+
+	// Deadline is the deadline value for the task.
+	// If task processing doesn't complete before the deadline, the task will be retried.
+	// The value time.Unix(0, 0) means no deadline.
+	//
+	// If timeout is set, min(now+timeout, deadline) is used, where the now is the time when
+	// a worker starts processing the task.
+	Deadline time.Time
+}
+
 // EnqueueAt schedules task to be enqueued at the specified time.
 //
 // EnqueueAt returns nil if the task is scheduled successfully, otherwise returns a non-nil error.
 //
 // The argument opts specifies the behavior of task processing.
 // If there are conflicting Option values the last one overrides others.
-func (c *Client) EnqueueAt(t time.Time, task *Task, opts ...Option) error {
+// 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) (*Result, error) {
 	return c.enqueueAt(t, task, opts...)
 }
 
@@ -195,7 +246,8 @@ 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.
-func (c *Client) Enqueue(task *Task, opts ...Option) error {
+// By deafult, max retry is set to 25 and timeout is set to 30 minutes.
+func (c *Client) Enqueue(task *Task, opts ...Option) (*Result, error) {
 	return c.enqueueAt(time.Now(), task, opts...)
 }
 
@@ -205,7 +257,8 @@ 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.
-func (c *Client) EnqueueIn(d time.Duration, task *Task, opts ...Option) error {
+// 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) (*Result, error) {
 	return c.enqueueAt(time.Now().Add(d), task, opts...)
 }
 
@@ -214,21 +267,33 @@ func (c *Client) Close() error {
 	return c.rdb.Close()
 }
 
-func (c *Client) enqueueAt(t time.Time, task *Task, opts ...Option) error {
+func (c *Client) enqueueAt(t time.Time, task *Task, opts ...Option) (*Result, error) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	if defaults, ok := c.opts[task.Type]; ok {
 		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(),
+		ID:        uuid.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
@@ -238,10 +303,19 @@ func (c *Client) enqueueAt(t time.Time, task *Task, opts ...Option) error {
 	} else {
 		err = c.schedule(msg, t, opt.uniqueTTL)
 	}
-	if err == rdb.ErrDuplicateTask {
-		return fmt.Errorf("%w", ErrDuplicateTask)
+	switch {
+	case err == rdb.ErrDuplicateTask:
+		return nil, fmt.Errorf("%w", ErrDuplicateTask)
+	case err != nil:
+		return nil, err
 	}
-	return err
+	return &Result{
+		ID:       msg.ID.String(),
+		Queue:    msg.Queue,
+		Retry:    msg.Retry,
+		Timeout:  timeout,
+		Deadline: deadline,
+	}, nil
 }
 
 func (c *Client) enqueue(msg *base.TaskMessage, uniqueTTL time.Duration) 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/doc.go b/doc.go
index aaaaa0c..8f8c87d 100644
--- a/doc.go
+++ b/doc.go
@@ -25,10 +25,10 @@ Task is created with two parameters: its type and payload.
         map[string]interface{}{"user_id": 42})
 
     // Enqueue the task to be processed immediately.
-    err := client.Enqueue(t)
+    res, err := client.Enqueue(t)
 
     // Schedule the task to be processed after one minute.
-    err = client.EnqueueIn(time.Minute, t)
+    res, err = client.EnqueueIn(time.Minute, t)
 
 The Server is used to run the background task processing with a given
 handler.
diff --git a/go.mod b/go.mod
index 02cba88..2857432 100644
--- a/go.mod
+++ b/go.mod
@@ -5,7 +5,7 @@ go 1.13
 require (
 	github.com/go-redis/redis/v7 v7.2.0
 	github.com/google/go-cmp v0.4.0
-	github.com/rs/xid v1.2.1
+	github.com/google/uuid v1.1.1
 	github.com/spf13/cast v1.3.1
 	go.uber.org/goleak v0.10.0
 	golang.org/x/sys v0.0.0-20191204072324-ce4227a45e2e
diff --git a/go.sum b/go.sum
index 913a295..28cc09f 100644
--- a/go.sum
+++ b/go.sum
@@ -2,16 +2,15 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/fsnotify/fsnotify v1.4.7 h1:IXs+QLmnXW2CcXuY+8Mzv/fWEsPGWxqefPtCP5CnV9I=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
-github.com/go-redis/redis/v7 v7.0.0-beta.4 h1:p6z7Pde69EGRWvlC++y8aFcaWegyrKHzOBGo0zUACTQ=
-github.com/go-redis/redis/v7 v7.0.0-beta.4/go.mod h1:xhhSbUMTsleRPur+Vgx9sUHtyN33bdjxY+9/0n9Ig8s=
 github.com/go-redis/redis/v7 v7.2.0 h1:CrCexy/jYWZjW0AyVoHlcJUeZN19VWlbepTh1Vq6dJs=
 github.com/go-redis/redis/v7 v7.2.0/go.mod h1:JDNMw23GTyLNC4GZu9njt15ctBQVn7xjRfnwdHj/Dcg=
 github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
-github.com/golang/protobuf v1.3.1 h1:YF8+flBXS5eO826T4nzqPrxfhQThhXl0YzfuUPu4SBg=
-github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
+github.com/golang/protobuf v1.3.2 h1:6nsPYzhq5kReh6QImI3k5qWzO4PEbvbIW2cwSfR/6xs=
 github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/google/go-cmp v0.4.0 h1:xsAVV57WRhGj6kEIi8ReJzQlHHqcBYCElAvkovg3B/4=
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/uuid v1.1.1 h1:Gkbcsh/GbpXz7lPftLA3P6TYMwjCLYm83jiFQZF/3gY=
+github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/hpcloud/tail v1.0.0 h1:nfCOvKYfkgYP8hkirhJocXT2+zOD8yUNjXaWfTlyFKI=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
 github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
@@ -20,16 +19,12 @@ github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
-github.com/onsi/ginkgo v1.8.0 h1:VkHVNpR4iVnU8XQR6DBm8BqYjN7CRzw+xKUbVVbbW9w=
-github.com/onsi/ginkgo v1.8.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
+github.com/onsi/ginkgo v1.10.1 h1:q/mM8GF/n0shIN8SaAZ0V+jnLPzen6WIVZdiwrRlMlo=
 github.com/onsi/ginkgo v1.10.1/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
-github.com/onsi/gomega v1.5.0 h1:izbySO9zDPmjJ8rDjLvkA2zJHIo+HkYXHnf7eN7SSyo=
-github.com/onsi/gomega v1.5.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
+github.com/onsi/gomega v1.7.0 h1:XPnZz8VVBHjVsy1vzJmRwIcSwiUO+JFfrv/xGiigmME=
 github.com/onsi/gomega v1.7.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/rs/xid v1.2.1 h1:mhH9Nq+C1fY2l1XIpgxIiUOfNpRBYH1kKcr+qfKgjRc=
-github.com/rs/xid v1.2.1/go.mod h1:+uKXf+4Djp6Md1KODXJxgGQPKngRmWyn10oCKFzNHOQ=
 github.com/spf13/cast v1.3.1 h1:nFm6S0SMdyzrzcmThSipiEubIDy8WEXKNZ0UOgiRpng=
 github.com/spf13/cast v1.3.1/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE=
 github.com/stretchr/testify v1.2.2 h1:bSDNvY7ZPG5RlJ8otE/7V6gMiyenm9RtJ7IUVIAoJ1w=
@@ -39,10 +34,8 @@ go.uber.org/goleak v0.10.0/go.mod h1:VCZuO8V8mFPlL0F5J5GK1rtHV3DrFcQ1R8ryq7FK0aI
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/net v0.0.0-20180906233101-161cd47e91fd h1:nTDtHvHSdCn1m6ITfMRqtOd/9+7a3s8RBNOZ3eYZzJA=
 golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
-golang.org/x/net v0.0.0-20190522155817-f3200d17e092 h1:4QSRKanuywn15aTZvI/mIDEgPQpswuFndXpOj3rKEco=
-golang.org/x/net v0.0.0-20190522155817-f3200d17e092/go.mod h1:HSz+uSET+XFnRR8LxR5pz3Of3rY3CfYBVs4xY44aLks=
+golang.org/x/net v0.0.0-20190923162816-aa69164e4478 h1:l5EDrHhldLYb3ZRHDUhXF7Om7MvYXnkV9/iQNo1lX6g=
 golang.org/x/net v0.0.0-20190923162816-aa69164e4478/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
-golang.org/x/net v0.0.0-20200202094626-16171245cfb2 h1:CCH4IOTTfewWjGOlSp+zGcjutRKlBEZQ6wTn8ozI/nI=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e h1:o3PsSEY8E4eXWkXrIP9YJALUkVZqzHJT5DOasTyn8Vs=
 golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -60,8 +53,7 @@ golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGm
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
-gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/fsnotify.v1 v1.4.7 h1:xOHLXZwVvI9hhs+cLKq5+I5onOuwQLhQwiu63xxlHs4=
 gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
diff --git a/heartbeat.go b/heartbeat.go
index 26907f3..d504541 100644
--- a/heartbeat.go
+++ b/heartbeat.go
@@ -9,9 +9,9 @@ import (
 	"sync"
 	"time"
 
+	"github.com/google/uuid"
 	"github.com/hibiken/asynq/internal/base"
 	"github.com/hibiken/asynq/internal/log"
-	"github.com/rs/xid"
 )
 
 // heartbeater is responsible for writing process info to redis periodically to
@@ -74,7 +74,7 @@ func newHeartbeater(params heartbeaterParams) *heartbeater {
 
 		host:           host,
 		pid:            os.Getpid(),
-		serverID:       xid.New().String(),
+		serverID:       uuid.New().String(),
 		concurrency:    params.concurrency,
 		queues:         params.queues,
 		strictPriority: params.strictPriority,
diff --git a/internal/base/base.go b/internal/base/base.go
index 4733aa8..85c28d0 100644
--- a/internal/base/base.go
+++ b/internal/base/base.go
@@ -14,9 +14,12 @@ import (
 	"time"
 
 	"github.com/go-redis/redis/v7"
-	"github.com/rs/xid"
+	"github.com/google/uuid"
 )
 
+// Version of asynq library and CLI.
+const Version = "0.10.0"
+
 // DefaultQueueName is the queue name used if none are specified by user.
 const DefaultQueueName = "default"
 
@@ -35,6 +38,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
 )
@@ -74,7 +78,7 @@ type TaskMessage struct {
 	Payload map[string]interface{}
 
 	// ID is a unique identifier for each task.
-	ID xid.ID
+	ID uuid.UUID
 
 	// Queue is a name this message should be enqueued to.
 	Queue string
@@ -88,18 +92,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 +253,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 +267,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/inspect.go b/internal/rdb/inspect.go
index 0ba139a..158d22a 100644
--- a/internal/rdb/inspect.go
+++ b/internal/rdb/inspect.go
@@ -12,8 +12,8 @@ import (
 	"time"
 
 	"github.com/go-redis/redis/v7"
+	"github.com/google/uuid"
 	"github.com/hibiken/asynq/internal/base"
-	"github.com/rs/xid"
 	"github.com/spf13/cast"
 )
 
@@ -53,7 +53,7 @@ type DailyStats struct {
 
 // EnqueuedTask is a task in a queue and is ready to be processed.
 type EnqueuedTask struct {
-	ID      xid.ID
+	ID      uuid.UUID
 	Type    string
 	Payload map[string]interface{}
 	Queue   string
@@ -61,14 +61,14 @@ type EnqueuedTask struct {
 
 // InProgressTask is a task that's currently being processed.
 type InProgressTask struct {
-	ID      xid.ID
+	ID      uuid.UUID
 	Type    string
 	Payload map[string]interface{}
 }
 
 // ScheduledTask is a task that's scheduled to be processed in the future.
 type ScheduledTask struct {
-	ID        xid.ID
+	ID        uuid.UUID
 	Type      string
 	Payload   map[string]interface{}
 	ProcessAt time.Time
@@ -78,7 +78,7 @@ type ScheduledTask struct {
 
 // RetryTask is a task that's in retry queue because worker failed to process the task.
 type RetryTask struct {
-	ID      xid.ID
+	ID      uuid.UUID
 	Type    string
 	Payload map[string]interface{}
 	// TODO(hibiken): add LastFailedAt time.Time
@@ -92,7 +92,7 @@ type RetryTask struct {
 
 // DeadTask is a task in that has exhausted all retries.
 type DeadTask struct {
-	ID           xid.ID
+	ID           uuid.UUID
 	Type         string
 	Payload      map[string]interface{}
 	LastFailedAt time.Time
@@ -446,7 +446,7 @@ func (r *RDB) ListDead(pgn Pagination) ([]*DeadTask, error) {
 // EnqueueDeadTask finds a task that matches the given id and score from dead queue
 // and enqueues it for processing. If a task that matches the id and score
 // does not exist, it returns ErrTaskNotFound.
-func (r *RDB) EnqueueDeadTask(id xid.ID, score int64) error {
+func (r *RDB) EnqueueDeadTask(id uuid.UUID, score int64) error {
 	n, err := r.removeAndEnqueue(base.DeadQueue, id.String(), float64(score))
 	if err != nil {
 		return err
@@ -460,7 +460,7 @@ func (r *RDB) EnqueueDeadTask(id xid.ID, score int64) error {
 // EnqueueRetryTask finds a task that matches the given id and score from retry queue
 // and enqueues it for processing. If a task that matches the id and score
 // does not exist, it returns ErrTaskNotFound.
-func (r *RDB) EnqueueRetryTask(id xid.ID, score int64) error {
+func (r *RDB) EnqueueRetryTask(id uuid.UUID, score int64) error {
 	n, err := r.removeAndEnqueue(base.RetryQueue, id.String(), float64(score))
 	if err != nil {
 		return err
@@ -474,7 +474,7 @@ func (r *RDB) EnqueueRetryTask(id xid.ID, score int64) error {
 // EnqueueScheduledTask finds a task that matches the given id and score from scheduled queue
 // and enqueues it for processing. If a task that matches the id and score does not
 // exist, it returns ErrTaskNotFound.
-func (r *RDB) EnqueueScheduledTask(id xid.ID, score int64) error {
+func (r *RDB) EnqueueScheduledTask(id uuid.UUID, score int64) error {
 	n, err := r.removeAndEnqueue(base.ScheduledQueue, id.String(), float64(score))
 	if err != nil {
 		return err
@@ -553,7 +553,7 @@ func (r *RDB) removeAndEnqueueAll(zset string) (int64, error) {
 // KillRetryTask finds a task that matches the given id and score from retry queue
 // and moves it to dead queue. If a task that maches the id and score does not exist,
 // it returns ErrTaskNotFound.
-func (r *RDB) KillRetryTask(id xid.ID, score int64) error {
+func (r *RDB) KillRetryTask(id uuid.UUID, score int64) error {
 	n, err := r.removeAndKill(base.RetryQueue, id.String(), float64(score))
 	if err != nil {
 		return err
@@ -567,7 +567,7 @@ func (r *RDB) KillRetryTask(id xid.ID, score int64) error {
 // KillScheduledTask finds a task that matches the given id and score from scheduled queue
 // and moves it to dead queue. If a task that maches the id and score does not exist,
 // it returns ErrTaskNotFound.
-func (r *RDB) KillScheduledTask(id xid.ID, score int64) error {
+func (r *RDB) KillScheduledTask(id uuid.UUID, score int64) error {
 	n, err := r.removeAndKill(base.ScheduledQueue, id.String(), float64(score))
 	if err != nil {
 		return err
@@ -660,21 +660,21 @@ func (r *RDB) removeAndKillAll(zset string) (int64, error) {
 // DeleteDeadTask finds a task that matches the given id and score from dead queue
 // and deletes it. If a task that matches the id and score does not exist,
 // it returns ErrTaskNotFound.
-func (r *RDB) DeleteDeadTask(id xid.ID, score int64) error {
+func (r *RDB) DeleteDeadTask(id uuid.UUID, score int64) error {
 	return r.deleteTask(base.DeadQueue, id.String(), float64(score))
 }
 
 // DeleteRetryTask finds a task that matches the given id and score from retry queue
 // and deletes it. If a task that matches the id and score does not exist,
 // it returns ErrTaskNotFound.
-func (r *RDB) DeleteRetryTask(id xid.ID, score int64) error {
+func (r *RDB) DeleteRetryTask(id uuid.UUID, score int64) error {
 	return r.deleteTask(base.RetryQueue, id.String(), float64(score))
 }
 
 // DeleteScheduledTask finds a task that matches the given id and score from
 // scheduled queue  and deletes it. If a task that matches the id and score
 //does not exist, it returns ErrTaskNotFound.
-func (r *RDB) DeleteScheduledTask(id xid.ID, score int64) error {
+func (r *RDB) DeleteScheduledTask(id uuid.UUID, score int64) error {
 	return r.deleteTask(base.ScheduledQueue, id.String(), float64(score))
 }
 
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..1614d41 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,40 +184,46 @@ 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()
 				p.cancelations.Delete(msg.ID.String())
 			}()
 
+			// check context before starting a worker goroutine.
+			select {
+			case <-ctx.Done():
+				// already canceled (e.g. deadline exceeded).
+				p.retryOrKill(ctx, msg, ctx.Err())
+				return
+			default:
+			}
+
 			resCh := make(chan error, 1)
-			task := NewTask(msg.Type, msg.Payload)
-			go func() { resCh <- perform(ctx, task, p.handler) }()
+			go func() {
+				resCh <- perform(ctx, NewTask(msg.Type, msg.Payload), 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.retryOrKill(ctx, msg, ctx.Err())
+				return
 			case resErr := <-resCh:
 				// Note: One of three things should happen.
 				// 1) Done  -> Removes the message from InProgress
 				// 2) Retry -> Removes the message from InProgress & Adds the message to Retry
 				// 3) Kill  -> Removes the message from InProgress & Adds the message to Dead
 				if resErr != nil {
-					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 +238,73 @@ 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 p.errHandler != nil {
+		p.errHandler.HandleError(ctx, NewTask(msg.Type, msg.Payload), err)
+	}
+	if msg.Retried >= msg.Retry {
+		p.logger.Warnf("Retry exhausted for task id=%s", msg.ID)
+		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) {
-	p.logger.Warnf("Retry exhausted for task id=%s", msg.ID)
+func (p *processor) kill(ctx context.Context, msg *base.TaskMessage, e error) {
 	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..3fc8998 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.
@@ -124,18 +125,18 @@ type Config struct {
 	ShutdownTimeout time.Duration
 }
 
-// An ErrorHandler handles errors returned by the task handler.
+// An ErrorHandler handles an error occured during task processing.
 type ErrorHandler interface {
-	HandleError(task *Task, err error, retried, maxRetry int)
+	HandleError(ctx context.Context, task *Task, err error)
 }
 
 // The ErrorHandlerFunc type is an adapter to allow the use of  ordinary functions as a ErrorHandler.
 // If f is a function with the appropriate signature, ErrorHandlerFunc(f) is a ErrorHandler that calls f.
-type ErrorHandlerFunc func(task *Task, err error, retried, maxRetry int)
+type ErrorHandlerFunc func(ctx context.Context, task *Task, err error)
 
-// HandleError calls fn(task, err, retried, maxRetry)
-func (fn ErrorHandlerFunc) HandleError(task *Task, err error, retried, maxRetry int) {
-	fn(task, err, retried, maxRetry)
+// HandleError calls fn(ctx, task, err)
+func (fn ErrorHandlerFunc) HandleError(ctx context.Context, task *Task, err error) {
+	fn(ctx, task, err)
 }
 
 // Logger supports logging at various log levels.
@@ -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/ls.go b/tools/asynq/cmd/ls.go
index 8c82231..ad0cfd2 100644
--- a/tools/asynq/cmd/ls.go
+++ b/tools/asynq/cmd/ls.go
@@ -13,8 +13,8 @@ import (
 	"time"
 
 	"github.com/go-redis/redis/v7"
+	"github.com/google/uuid"
 	"github.com/hibiken/asynq/internal/rdb"
-	"github.com/rs/xid"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
 )
@@ -93,7 +93,7 @@ func ls(cmd *cobra.Command, args []string) {
 // queryID returns an identifier used for "enq" command.
 // score is the zset score and queryType should be one
 // of "s", "r" or "d" (scheduled, retry, dead respectively).
-func queryID(id xid.ID, score int64, qtype string) string {
+func queryID(id uuid.UUID, score int64, qtype string) string {
 	const format = "%v:%v:%v"
 	return fmt.Sprintf(format, qtype, score, id)
 }
@@ -101,22 +101,22 @@ func queryID(id xid.ID, score int64, qtype string) string {
 // parseQueryID is a reverse operation of queryID function.
 // It takes a queryID and return each part of id with proper
 // type if valid, otherwise it reports an error.
-func parseQueryID(queryID string) (id xid.ID, score int64, qtype string, err error) {
+func parseQueryID(queryID string) (id uuid.UUID, score int64, qtype string, err error) {
 	parts := strings.Split(queryID, ":")
 	if len(parts) != 3 {
-		return xid.NilID(), 0, "", fmt.Errorf("invalid id")
+		return uuid.Nil, 0, "", fmt.Errorf("invalid id")
 	}
-	id, err = xid.FromString(parts[2])
+	id, err = uuid.Parse(parts[2])
 	if err != nil {
-		return xid.NilID(), 0, "", fmt.Errorf("invalid id")
+		return uuid.Nil, 0, "", fmt.Errorf("invalid id")
 	}
 	score, err = strconv.ParseInt(parts[1], 10, 64)
 	if err != nil {
-		return xid.NilID(), 0, "", fmt.Errorf("invalid id")
+		return uuid.Nil, 0, "", fmt.Errorf("invalid id")
 	}
 	qtype = parts[0]
 	if len(qtype) != 1 || !strings.Contains("srd", qtype) {
-		return xid.NilID(), 0, "", fmt.Errorf("invalid id")
+		return uuid.Nil, 0, "", fmt.Errorf("invalid id")
 	}
 	return id, score, qtype, nil
 }
diff --git a/tools/asynq/cmd/migrate.go b/tools/asynq/cmd/migrate.go
new file mode 100644
index 0000000..7ed04c0
--- /dev/null
+++ b/tools/asynq/cmd/migrate.go
@@ -0,0 +1,212 @@
+// 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/google/uuid"
+	"github.com/hibiken/asynq/internal/base"
+	"github.com/spf13/cast"
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+)
+
+// migrateCmd represents the migrate command
+var migrateCmd = &cobra.Command{
+	Use:   "migrate",
+	Short: fmt.Sprintf("Migrate all tasks to be compatible with asynq@%s", base.Version),
+	Long:  fmt.Sprintf("Migrate (asynq migrate) will convert all tasks in redis to be compatible with asynq@%s.", base.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        uuid.UUID
+	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:        uuid.New(),
+		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/asynq/cmd/root.go b/tools/asynq/cmd/root.go
index 8056ac5..12ceaa4 100644
--- a/tools/asynq/cmd/root.go
+++ b/tools/asynq/cmd/root.go
@@ -11,6 +11,7 @@ import (
 	"strings"
 	"text/tabwriter"
 
+	"github.com/hibiken/asynq/internal/base"
 	"github.com/spf13/cobra"
 
 	homedir "github.com/mitchellh/go-homedir"
@@ -26,9 +27,20 @@ var password string
 
 // rootCmd represents the base command when called without any subcommands
 var rootCmd = &cobra.Command{
-	Use:   "asynq",
-	Short: "A monitoring tool for asynq queues",
-	Long:  `Asynq is a montoring CLI to inspect tasks and queues managed by asynq.`,
+	Use:     "asynq",
+	Short:   "A monitoring tool for asynq queues",
+	Long:    `Asynq is a montoring CLI to inspect tasks and queues managed by asynq.`,
+	Version: base.Version,
+}
+
+var versionOutput = fmt.Sprintf("asynq version %s\n", base.Version)
+
+var versionCmd = &cobra.Command{
+	Use:    "version",
+	Hidden: true,
+	Run: func(cmd *cobra.Command, args []string) {
+		fmt.Print(versionOutput)
+	},
 }
 
 // Execute adds all child commands to the root command and sets flags appropriately.
@@ -43,6 +55,9 @@ func Execute() {
 func init() {
 	cobra.OnInitialize(initConfig)
 
+	rootCmd.AddCommand(versionCmd)
+	rootCmd.SetVersionTemplate(versionOutput)
+
 	rootCmd.PersistentFlags().StringVar(&cfgFile, "config", "", "config file to set flag defaut values (default is $HOME/.asynq.yaml)")
 	rootCmd.PersistentFlags().StringVarP(&uri, "uri", "u", "127.0.0.1:6379", "redis server URI")
 	rootCmd.PersistentFlags().IntVarP(&db, "db", "n", 0, "redis database number (default is 0)")
diff --git a/tools/go.mod b/tools/go.mod
index 9685df1..fb60511 100644
--- a/tools/go.mod
+++ b/tools/go.mod
@@ -4,9 +4,10 @@ go 1.13
 
 require (
 	github.com/go-redis/redis/v7 v7.2.0
+	github.com/google/uuid v1.1.1
 	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
 )
diff --git a/tools/go.sum b/tools/go.sum
index 40205a1..537848c 100644
--- a/tools/go.sum
+++ b/tools/go.sum
@@ -1,4 +1,5 @@
 cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
+github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
@@ -15,6 +16,7 @@ github.com/coreos/go-semver v0.2.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3Ee
 github.com/coreos/go-systemd v0.0.0-20190321100706-95778dfbb74e/go.mod h1:F5haX7vjVVG0kc13fIWeqUViNPyEJxv/OmvnBo0Yme4=
 github.com/coreos/pkg v0.0.0-20180928190104-399ea9e2e55f/go.mod h1:E3G3o1h8I7cfcXa63jLwjI0eiQQMgzzUDFVpN/nH/eA=
 github.com/cpuguy83/go-md2man v1.0.10/go.mod h1:SmD6nW6nTyfqj6ABTjUi3V3JVMnlJmwcJI5acqYI6dE=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
 github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954/go.mod h1:vAd38F8PWV+bWy6jNmig1y/TA+kYO4g3RSRF0IAv0no=
@@ -24,10 +26,6 @@ github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeME
 github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
 github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9GBnD5lWE=
 github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk=
-github.com/go-redis/redis v6.15.7+incompatible h1:3skhDh95XQMpnqeqNftPkQD9jL9e5e36z/1SUm6dy1U=
-github.com/go-redis/redis/v7 v7.0.0-beta.4/go.mod h1:xhhSbUMTsleRPur+Vgx9sUHtyN33bdjxY+9/0n9Ig8s=
-github.com/go-redis/redis/v7 v7.1.0 h1:I4C4a8UGbFejiVjtYVTRVOiMIJ5pm5Yru6ibvDX/OS0=
-github.com/go-redis/redis/v7 v7.1.0/go.mod h1:JDNMw23GTyLNC4GZu9njt15ctBQVn7xjRfnwdHj/Dcg=
 github.com/go-redis/redis/v7 v7.2.0 h1:CrCexy/jYWZjW0AyVoHlcJUeZN19VWlbepTh1Vq6dJs=
 github.com/go-redis/redis/v7 v7.2.0/go.mod h1:JDNMw23GTyLNC4GZu9njt15ctBQVn7xjRfnwdHj/Dcg=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
@@ -38,10 +36,15 @@ github.com/golang/groupcache v0.0.0-20190129154638-5b532d6fd5ef/go.mod h1:cIg4er
 github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
 github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
+github.com/golang/protobuf v1.3.2 h1:6nsPYzhq5kReh6QImI3k5qWzO4PEbvbIW2cwSfR/6xs=
 github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
+github.com/google/go-cmp v0.4.0 h1:xsAVV57WRhGj6kEIi8ReJzQlHHqcBYCElAvkovg3B/4=
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/uuid v1.1.1 h1:Gkbcsh/GbpXz7lPftLA3P6TYMwjCLYm83jiFQZF/3gY=
+github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1 h1:EGx4pi6eqNxGaHF6qqu48+N2wcFQ5qg5FXgOdqsJ5d8=
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
 github.com/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ=
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.0/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs=
@@ -49,19 +52,22 @@ github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0/go.mod h1:8NvIoxWQoOIhqOTXgf
 github.com/grpc-ecosystem/grpc-gateway v1.9.0/go.mod h1:vNeuVxBJEsws4ogUvrchl83t/GYV9WGTSLVdBhOQFDY=
 github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4=
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
-github.com/hibiken/asynq v0.4.0 h1:NvAfYX0DRe04WgGMKRg5oX7bs6ktv2fu9YwB6O356FI=
-github.com/hibiken/asynq v0.4.0/go.mod h1:dtrVkxCsGPVhVNHMDXAH7lFq64kbj43+G6lt4FQZfW4=
+github.com/hpcloud/tail v1.0.0 h1:nfCOvKYfkgYP8hkirhJocXT2+zOD8yUNjXaWfTlyFKI=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
+github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NHg9XEKhtSvM=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
 github.com/jonboulle/clockwork v0.1.0/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22gdxWY5EU2bo=
+github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7C0MuV77Wo=
 github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
 github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
 github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvWXihfKN4Q=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
+github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
+github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/magiconair/properties v1.8.0/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
 github.com/magiconair/properties v1.8.1 h1:ZC2Vc7/ZFkGmsVC9KvOjumD+G5lXy2RtTKyzRKO2BQ4=
@@ -74,15 +80,14 @@ github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh
 github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
 github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U=
 github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
-github.com/onsi/ginkgo v1.8.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
+github.com/onsi/ginkgo v1.10.1 h1:q/mM8GF/n0shIN8SaAZ0V+jnLPzen6WIVZdiwrRlMlo=
 github.com/onsi/ginkgo v1.10.1/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
-github.com/onsi/gomega v1.5.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
+github.com/onsi/gomega v1.7.0 h1:XPnZz8VVBHjVsy1vzJmRwIcSwiUO+JFfrv/xGiigmME=
 github.com/onsi/gomega v1.7.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
 github.com/pelletier/go-toml v1.2.0 h1:T5zMGML61Wp+FlcbWjRDT7yAxhJNAiPPLOFECq181zc=
 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
-github.com/pelletier/go-toml v1.6.0 h1:aetoXYr0Tv7xRU/V4B4IZJ2QcbtMUFoNb3ORp7TzIK4=
-github.com/pelletier/go-toml v1.6.0/go.mod h1:5N711Q9dKgbdkxHL+MEfF31hpT7l0S0s/t2kKREewys=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
 github.com/prometheus/client_golang v0.9.3/go.mod h1:/TN21ttK/J9q6uSwhBd54HahCDft0ttaMvbicHlPoso=
@@ -94,18 +99,16 @@ github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R
 github.com/prometheus/procfs v0.0.0-20190507164030-5867b95ac084/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA=
 github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU=
 github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6SoW27p1b0cqNHllgS5HIMJraePCO15w5zCzIWYg=
-github.com/rs/xid v1.2.1 h1:mhH9Nq+C1fY2l1XIpgxIiUOfNpRBYH1kKcr+qfKgjRc=
-github.com/rs/xid v1.2.1/go.mod h1:+uKXf+4Djp6Md1KODXJxgGQPKngRmWyn10oCKFzNHOQ=
 github.com/russross/blackfriday v1.5.2/go.mod h1:JO/DiYxRf+HjHt06OyowR9PTA263kcR/rfWxYHBV53g=
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
+github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d h1:zE9ykElWQ6/NYmHa3jpm/yHnI4xSofP+UP6SpjHcSeM=
 github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc=
+github.com/smartystreets/goconvey v1.6.4 h1:fv0U8FUIMPNf1L9lnHLvLhgicrIVChEkdzIKYqbNC9s=
 github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA=
 github.com/soheilhy/cmux v0.1.4/go.mod h1:IM3LyeVVIOuxMH7sFAkER9+bJ4dT7Ms6E4xg4kGIyLM=
 github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
 github.com/spf13/afero v1.1.2 h1:m8/z1t7/fwjysjQRYbP0RD+bUIF/8tJwPdEZsI83ACI=
 github.com/spf13/afero v1.1.2/go.mod h1:j4pytiNVoe2o6bmDsKpLACNPDBIoEAkihy7loJ1B0CQ=
-github.com/spf13/afero v1.2.2 h1:5jhuqJyZCZf2JRofRvN/nIFgIWNzPa3/Vz8mYylgbWc=
-github.com/spf13/afero v1.2.2/go.mod h1:9ZxEEn6pIJ8Rxe320qSDBk6AsU0r9pR7Q4OcevTdifk=
 github.com/spf13/cast v1.3.0/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE=
 github.com/spf13/cast v1.3.1 h1:nFm6S0SMdyzrzcmThSipiEubIDy8WEXKNZ0UOgiRpng=
 github.com/spf13/cast v1.3.1/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE=
@@ -113,17 +116,13 @@ github.com/spf13/cobra v0.0.5 h1:f0B+LkLX6DtmRH1isoNA9VTtNUK9K8xYd28JNNfOv/s=
 github.com/spf13/cobra v0.0.5/go.mod h1:3K3wKZymM7VvHMDS9+Akkh4K60UwM26emMESw8tLCHU=
 github.com/spf13/jwalterweatherman v1.0.0 h1:XHEdyB+EcvlqZamSM4ZOMGlc93t6AcsBEu9Gc1vn7yk=
 github.com/spf13/jwalterweatherman v1.0.0/go.mod h1:cQK4TGJAtQXfYWX+Ddv3mKDzgVb68N+wFjFa4jdeBTo=
-github.com/spf13/jwalterweatherman v1.1.0 h1:ue6voC5bR5F8YxI5S67j9i582FU4Qvo2bmqnqMYADFk=
-github.com/spf13/jwalterweatherman v1.1.0/go.mod h1:aNWZUN0dPAAO/Ljvb5BEdw96iTZ0EXowPYD95IqWIGo=
 github.com/spf13/pflag v1.0.3 h1:zPAT6CGy6wXeQ7NtTnaTerfKOsV6V6F8agHXFiazDkg=
 github.com/spf13/pflag v1.0.3/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4=
-github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA=
-github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
 github.com/spf13/viper v1.3.2/go.mod h1:ZiWeW+zYFKm7srdB9IoDzzZXaJaI5eL9QjNiN/DMA2s=
-github.com/spf13/viper v1.6.0/go.mod h1:t3iDnF5Jlj76alVNuyFBk5oUMCvsrkbvZK0WQdfDi5k=
 github.com/spf13/viper v1.6.2 h1:7aKfF+e8/k68gda3LOjo5RxiUqddoFxVq4BKBPrxk5E=
 github.com/spf13/viper v1.6.2/go.mod h1:t3iDnF5Jlj76alVNuyFBk5oUMCvsrkbvZK0WQdfDi5k=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.2.2 h1:bSDNvY7ZPG5RlJ8otE/7V6gMiyenm9RtJ7IUVIAoJ1w=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/subosito/gotenv v1.2.0 h1:Slr1R9HxAlEKefgq5jn9U+DnETlIUa6HfgEzj0g5d7s=
 github.com/subosito/gotenv v1.2.0/go.mod h1:N0PQaV/YGNqwC0u51sEeR/aUtSLEXKX9iv69rRypqCw=
@@ -148,6 +147,7 @@ golang.org/x/net v0.0.0-20181114220301-adae6a3d119a/go.mod h1:mL1N/T3taQHkDXs73r
 golang.org/x/net v0.0.0-20181220203305-927f97764cc3/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190522155817-f3200d17e092/go.mod h1:HSz+uSET+XFnRR8LxR5pz3Of3rY3CfYBVs4xY44aLks=
+golang.org/x/net v0.0.0-20190923162816-aa69164e4478 h1:l5EDrHhldLYb3ZRHDUhXF7Om7MvYXnkV9/iQNo1lX6g=
 golang.org/x/net v0.0.0-20190923162816-aa69164e4478/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -172,6 +172,7 @@ golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGm
 golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20190328211700-ab21143f2384/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
 google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc=
@@ -180,11 +181,14 @@ google.golang.org/grpc v1.21.0/go.mod h1:oYelfM1adQP15Ek0mdvEgi9Df8B9CZIaU1084ij
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/fsnotify.v1 v1.4.7 h1:xOHLXZwVvI9hhs+cLKq5+I5onOuwQLhQwiu63xxlHs4=
 gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
 gopkg.in/ini.v1 v1.51.0 h1:AQvPpx3LzTDM0AjnIRlVFwFFGC+npRopjZxLJj6gdno=
 gopkg.in/ini.v1 v1.51.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=
 gopkg.in/resty.v1 v1.12.0/go.mod h1:mDo4pnntr5jdWRML875a/NmxYqAlA73dVijT2AXvQQo=
+gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ=
 gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw=
 gopkg.in/yaml.v2 v2.0.0-20170812160011-eb3733d160e7/go.mod h1:JAlM8MvJe8wmxCU4Bli9HhUf9+ttbYbLASfIpnQbh74=
 gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=


Test patch:
diff --git a/benchmark_test.go b/benchmark_test.go
index 59b9e7e..df2d190 100644
--- a/benchmark_test.go
+++ b/benchmark_test.go
@@ -33,7 +33,7 @@ func BenchmarkEndToEndSimple(b *testing.B) {
 		// Create a bunch of tasks
 		for i := 0; i < count; i++ {
 			t := NewTask(fmt.Sprintf("task%d", i), map[string]interface{}{"data": i})
-			if err := client.Enqueue(t); err != nil {
+			if _, err := client.Enqueue(t); err != nil {
 				b.Fatalf("could not enqueue a task: %v", err)
 			}
 		}
@@ -76,13 +76,13 @@ func BenchmarkEndToEnd(b *testing.B) {
 		// Create a bunch of tasks
 		for i := 0; i < count; i++ {
 			t := NewTask(fmt.Sprintf("task%d", i), map[string]interface{}{"data": i})
-			if err := client.Enqueue(t); err != nil {
+			if _, err := client.Enqueue(t); err != nil {
 				b.Fatalf("could not enqueue a task: %v", err)
 			}
 		}
 		for i := 0; i < count; i++ {
 			t := NewTask(fmt.Sprintf("scheduled%d", i), map[string]interface{}{"data": i})
-			if err := client.EnqueueAt(time.Now().Add(time.Second), t); err != nil {
+			if _, err := client.EnqueueAt(time.Now().Add(time.Second), t); err != nil {
 				b.Fatalf("could not enqueue a task: %v", err)
 			}
 		}
@@ -144,19 +144,19 @@ func BenchmarkEndToEndMultipleQueues(b *testing.B) {
 		// Create a bunch of tasks
 		for i := 0; i < highCount; i++ {
 			t := NewTask(fmt.Sprintf("task%d", i), map[string]interface{}{"data": i})
-			if err := client.Enqueue(t, Queue("high")); err != nil {
+			if _, err := client.Enqueue(t, Queue("high")); err != nil {
 				b.Fatalf("could not enqueue a task: %v", err)
 			}
 		}
 		for i := 0; i < defaultCount; i++ {
 			t := NewTask(fmt.Sprintf("task%d", i), map[string]interface{}{"data": i})
-			if err := client.Enqueue(t); err != nil {
+			if _, err := client.Enqueue(t); err != nil {
 				b.Fatalf("could not enqueue a task: %v", err)
 			}
 		}
 		for i := 0; i < lowCount; i++ {
 			t := NewTask(fmt.Sprintf("task%d", i), map[string]interface{}{"data": i})
-			if err := client.Enqueue(t, Queue("low")); err != nil {
+			if _, err := client.Enqueue(t, Queue("low")); err != nil {
 				b.Fatalf("could not enqueue a task: %v", err)
 			}
 		}
@@ -200,14 +200,14 @@ func BenchmarkClientWhileServerRunning(b *testing.B) {
 		// Enqueue 10,000 tasks.
 		for i := 0; i < count; i++ {
 			t := NewTask(fmt.Sprintf("task%d", i), map[string]interface{}{"data": i})
-			if err := client.Enqueue(t); err != nil {
+			if _, err := client.Enqueue(t); err != nil {
 				b.Fatalf("could not enqueue a task: %v", err)
 			}
 		}
 		// Schedule 10,000 tasks.
 		for i := 0; i < count; i++ {
 			t := NewTask(fmt.Sprintf("scheduled%d", i), map[string]interface{}{"data": i})
-			if err := client.EnqueueAt(time.Now().Add(time.Second), t); err != nil {
+			if _, err := client.EnqueueAt(time.Now().Add(time.Second), t); err != nil {
 				b.Fatalf("could not enqueue a task: %v", err)
 			}
 		}
@@ -223,7 +223,7 @@ func BenchmarkClientWhileServerRunning(b *testing.B) {
 		enqueued := 0
 		for enqueued < 100000 {
 			t := NewTask(fmt.Sprintf("enqueued%d", enqueued), map[string]interface{}{"data": enqueued})
-			if err := client.Enqueue(t); err != nil {
+			if _, err := client.Enqueue(t); err != nil {
 				b.Logf("could not enqueue task %d: %v", enqueued, err)
 				continue
 			}
diff --git a/client_test.go b/client_test.go
index 348d056..c461dff 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{
@@ -39,6 +34,7 @@ func TestClientEnqueueAt(t *testing.T) {
 		task          *Task
 		processAt     time.Time
 		opts          []Option
+		wantRes       *Result
 		wantEnqueued  map[string][]*base.TaskMessage
 		wantScheduled []h.ZSetEntry
 	}{
@@ -47,6 +43,12 @@ func TestClientEnqueueAt(t *testing.T) {
 			task:      task,
 			processAt: now,
 			opts:      []Option{},
+			wantRes: &Result{
+				Queue:    "default",
+				Retry:    defaultMaxRetry,
+				Timeout:  defaultTimeout,
+				Deadline: noDeadline,
+			},
 			wantEnqueued: map[string][]*base.TaskMessage{
 				"default": {
 					{
@@ -54,18 +56,24 @@ func TestClientEnqueueAt(t *testing.T) {
 						Payload:  task.Payload.data,
 						Retry:    defaultMaxRetry,
 						Queue:    "default",
-						Timeout:  noTimeout,
-						Deadline: noDeadline,
+						Timeout:  int64(defaultTimeout.Seconds()),
+						Deadline: noDeadline.Unix(),
 					},
 				},
 			},
 			wantScheduled: nil, // db is flushed in setup so zset does not exist hence nil
 		},
 		{
-			desc:         "Schedule task to be processed in the future",
-			task:         task,
-			processAt:    oneHourLater,
-			opts:         []Option{},
+			desc:      "Schedule task to be processed in the future",
+			task:      task,
+			processAt: oneHourLater,
+			opts:      []Option{},
+			wantRes: &Result{
+				Queue:    "default",
+				Retry:    defaultMaxRetry,
+				Timeout:  defaultTimeout,
+				Deadline: noDeadline,
+			},
 			wantEnqueued: nil, // db is flushed in setup so list does not exist hence nil
 			wantScheduled: []h.ZSetEntry{
 				{
@@ -74,8 +82,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()),
 				},
@@ -86,11 +94,15 @@ func TestClientEnqueueAt(t *testing.T) {
 	for _, tc := range tests {
 		h.FlushDB(t, r) // clean up db before each test case.
 
-		err := client.EnqueueAt(tc.processAt, tc.task, tc.opts...)
+		gotRes, err := client.EnqueueAt(tc.processAt, tc.task, tc.opts...)
 		if err != nil {
 			t.Error(err)
 			continue
 		}
+		if diff := cmp.Diff(tc.wantRes, gotRes, cmpopts.IgnoreFields(Result{}, "ID")); diff != "" {
+			t.Errorf("%s;\nEnqueueAt(processAt, task) returned %v, want %v; (-want,+got)\n%s",
+				tc.desc, gotRes, tc.wantRes, diff)
+		}
 
 		for qname, want := range tc.wantEnqueued {
 			gotEnqueued := h.GetEnqueuedMessages(t, r, qname)
@@ -119,6 +131,7 @@ func TestClientEnqueue(t *testing.T) {
 		desc         string
 		task         *Task
 		opts         []Option
+		wantRes      *Result
 		wantEnqueued map[string][]*base.TaskMessage
 	}{
 		{
@@ -127,6 +140,12 @@ func TestClientEnqueue(t *testing.T) {
 			opts: []Option{
 				MaxRetry(3),
 			},
+			wantRes: &Result{
+				Queue:    "default",
+				Retry:    3,
+				Timeout:  defaultTimeout,
+				Deadline: noDeadline,
+			},
 			wantEnqueued: map[string][]*base.TaskMessage{
 				"default": {
 					{
@@ -134,8 +153,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(),
 					},
 				},
 			},
@@ -146,6 +165,12 @@ func TestClientEnqueue(t *testing.T) {
 			opts: []Option{
 				MaxRetry(-2),
 			},
+			wantRes: &Result{
+				Queue:    "default",
+				Retry:    0,
+				Timeout:  defaultTimeout,
+				Deadline: noDeadline,
+			},
 			wantEnqueued: map[string][]*base.TaskMessage{
 				"default": {
 					{
@@ -153,8 +178,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(),
 					},
 				},
 			},
@@ -166,6 +191,12 @@ func TestClientEnqueue(t *testing.T) {
 				MaxRetry(2),
 				MaxRetry(10),
 			},
+			wantRes: &Result{
+				Queue:    "default",
+				Retry:    10,
+				Timeout:  defaultTimeout,
+				Deadline: noDeadline,
+			},
 			wantEnqueued: map[string][]*base.TaskMessage{
 				"default": {
 					{
@@ -173,8 +204,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(),
 					},
 				},
 			},
@@ -185,6 +216,12 @@ func TestClientEnqueue(t *testing.T) {
 			opts: []Option{
 				Queue("custom"),
 			},
+			wantRes: &Result{
+				Queue:    "custom",
+				Retry:    defaultMaxRetry,
+				Timeout:  defaultTimeout,
+				Deadline: noDeadline,
+			},
 			wantEnqueued: map[string][]*base.TaskMessage{
 				"custom": {
 					{
@@ -192,8 +229,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(),
 					},
 				},
 			},
@@ -204,6 +241,12 @@ func TestClientEnqueue(t *testing.T) {
 			opts: []Option{
 				Queue("HIGH"),
 			},
+			wantRes: &Result{
+				Queue:    "high",
+				Retry:    defaultMaxRetry,
+				Timeout:  defaultTimeout,
+				Deadline: noDeadline,
+			},
 			wantEnqueued: map[string][]*base.TaskMessage{
 				"high": {
 					{
@@ -211,8 +254,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(),
 					},
 				},
 			},
@@ -223,6 +266,12 @@ func TestClientEnqueue(t *testing.T) {
 			opts: []Option{
 				Timeout(20 * time.Second),
 			},
+			wantRes: &Result{
+				Queue:    "default",
+				Retry:    defaultMaxRetry,
+				Timeout:  20 * time.Second,
+				Deadline: noDeadline,
+			},
 			wantEnqueued: map[string][]*base.TaskMessage{
 				"default": {
 					{
@@ -230,8 +279,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(),
 					},
 				},
 			},
@@ -242,6 +291,12 @@ func TestClientEnqueue(t *testing.T) {
 			opts: []Option{
 				Deadline(time.Date(2020, time.June, 24, 0, 0, 0, 0, time.UTC)),
 			},
+			wantRes: &Result{
+				Queue:    "default",
+				Retry:    defaultMaxRetry,
+				Timeout:  noTimeout,
+				Deadline: time.Date(2020, time.June, 24, 0, 0, 0, 0, time.UTC),
+			},
 			wantEnqueued: map[string][]*base.TaskMessage{
 				"default": {
 					{
@@ -249,8 +304,34 @@ 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)),
+			},
+			wantRes: &Result{
+				Queue:    "default",
+				Retry:    defaultMaxRetry,
+				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(),
 					},
 				},
 			},
@@ -260,11 +341,15 @@ func TestClientEnqueue(t *testing.T) {
 	for _, tc := range tests {
 		h.FlushDB(t, r) // clean up db before each test case.
 
-		err := client.Enqueue(tc.task, tc.opts...)
+		gotRes, err := client.Enqueue(tc.task, tc.opts...)
 		if err != nil {
 			t.Error(err)
 			continue
 		}
+		if diff := cmp.Diff(tc.wantRes, gotRes, cmpopts.IgnoreFields(Result{}, "ID")); diff != "" {
+			t.Errorf("%s;\nEnqueue(task) returned %v, want %v; (-want,+got)\n%s",
+				tc.desc, gotRes, tc.wantRes, diff)
+		}
 
 		for qname, want := range tc.wantEnqueued {
 			got := h.GetEnqueuedMessages(t, r, qname)
@@ -289,14 +374,21 @@ func TestClientEnqueueIn(t *testing.T) {
 		task          *Task
 		delay         time.Duration
 		opts          []Option
+		wantRes       *Result
 		wantEnqueued  map[string][]*base.TaskMessage
 		wantScheduled []h.ZSetEntry
 	}{
 		{
-			desc:         "schedule a task to be enqueued in one hour",
-			task:         task,
-			delay:        time.Hour,
-			opts:         []Option{},
+			desc:  "schedule a task to be enqueued in one hour",
+			task:  task,
+			delay: time.Hour,
+			opts:  []Option{},
+			wantRes: &Result{
+				Queue:    "default",
+				Retry:    defaultMaxRetry,
+				Timeout:  defaultTimeout,
+				Deadline: noDeadline,
+			},
 			wantEnqueued: nil, // db is flushed in setup so list does not exist hence nil
 			wantScheduled: []h.ZSetEntry{
 				{
@@ -305,8 +397,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()),
 				},
@@ -317,6 +409,12 @@ func TestClientEnqueueIn(t *testing.T) {
 			task:  task,
 			delay: 0,
 			opts:  []Option{},
+			wantRes: &Result{
+				Queue:    "default",
+				Retry:    defaultMaxRetry,
+				Timeout:  defaultTimeout,
+				Deadline: noDeadline,
+			},
 			wantEnqueued: map[string][]*base.TaskMessage{
 				"default": {
 					{
@@ -324,8 +422,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(),
 					},
 				},
 			},
@@ -336,11 +434,15 @@ func TestClientEnqueueIn(t *testing.T) {
 	for _, tc := range tests {
 		h.FlushDB(t, r) // clean up db before each test case.
 
-		err := client.EnqueueIn(tc.delay, tc.task, tc.opts...)
+		gotRes, err := client.EnqueueIn(tc.delay, tc.task, tc.opts...)
 		if err != nil {
 			t.Error(err)
 			continue
 		}
+		if diff := cmp.Diff(tc.wantRes, gotRes, cmpopts.IgnoreFields(Result{}, "ID")); diff != "" {
+			t.Errorf("%s;\nEnqueueIn(delay, task) returned %v, want %v; (-want,+got)\n%s",
+				tc.desc, gotRes, tc.wantRes, diff)
+		}
 
 		for qname, want := range tc.wantEnqueued {
 			gotEnqueued := h.GetEnqueuedMessages(t, r, qname)
@@ -364,6 +466,7 @@ func TestClientDefaultOptions(t *testing.T) {
 		defaultOpts []Option // options set at the client level.
 		opts        []Option // options used at enqueue time.
 		task        *Task
+		wantRes     *Result
 		queue       string // queue that the message should go into.
 		want        *base.TaskMessage
 	}{
@@ -372,14 +475,20 @@ func TestClientDefaultOptions(t *testing.T) {
 			defaultOpts: []Option{Queue("feed")},
 			opts:        []Option{},
 			task:        NewTask("feed:import", nil),
-			queue:       "feed",
+			wantRes: &Result{
+				Queue:    "feed",
+				Retry:    defaultMaxRetry,
+				Timeout:  defaultTimeout,
+				Deadline: noDeadline,
+			},
+			queue: "feed",
 			want: &base.TaskMessage{
 				Type:     "feed:import",
 				Payload:  nil,
 				Retry:    defaultMaxRetry,
 				Queue:    "feed",
-				Timeout:  noTimeout,
-				Deadline: noDeadline,
+				Timeout:  int64(defaultTimeout.Seconds()),
+				Deadline: noDeadline.Unix(),
 			},
 		},
 		{
@@ -387,14 +496,20 @@ func TestClientDefaultOptions(t *testing.T) {
 			defaultOpts: []Option{Queue("feed"), MaxRetry(5)},
 			opts:        []Option{},
 			task:        NewTask("feed:import", nil),
-			queue:       "feed",
+			wantRes: &Result{
+				Queue:    "feed",
+				Retry:    5,
+				Timeout:  defaultTimeout,
+				Deadline: noDeadline,
+			},
+			queue: "feed",
 			want: &base.TaskMessage{
 				Type:     "feed:import",
 				Payload:  nil,
 				Retry:    5,
 				Queue:    "feed",
-				Timeout:  noTimeout,
-				Deadline: noDeadline,
+				Timeout:  int64(defaultTimeout.Seconds()),
+				Deadline: noDeadline.Unix(),
 			},
 		},
 		{
@@ -402,14 +517,20 @@ func TestClientDefaultOptions(t *testing.T) {
 			defaultOpts: []Option{Queue("feed"), MaxRetry(5)},
 			opts:        []Option{Queue("critical")},
 			task:        NewTask("feed:import", nil),
-			queue:       "critical",
+			wantRes: &Result{
+				Queue:    "critical",
+				Retry:    5,
+				Timeout:  defaultTimeout,
+				Deadline: noDeadline,
+			},
+			queue: "critical",
 			want: &base.TaskMessage{
 				Type:     "feed:import",
 				Payload:  nil,
 				Retry:    5,
 				Queue:    "critical",
-				Timeout:  noTimeout,
-				Deadline: noDeadline,
+				Timeout:  int64(defaultTimeout.Seconds()),
+				Deadline: noDeadline.Unix(),
 			},
 		},
 	}
@@ -418,10 +539,14 @@ func TestClientDefaultOptions(t *testing.T) {
 		h.FlushDB(t, r)
 		c := NewClient(RedisClientOpt{Addr: redisAddr, DB: redisDB})
 		c.SetDefaultOptions(tc.task.Type, tc.defaultOpts...)
-		err := c.Enqueue(tc.task, tc.opts...)
+		gotRes, err := c.Enqueue(tc.task, tc.opts...)
 		if err != nil {
 			t.Fatal(err)
 		}
+		if diff := cmp.Diff(tc.wantRes, gotRes, cmpopts.IgnoreFields(Result{}, "ID")); diff != "" {
+			t.Errorf("%s;\nEnqueue(task, opts...) returned %v, want %v; (-want,+got)\n%s",
+				tc.desc, gotRes, tc.wantRes, diff)
+		}
 		enqueued := h.GetEnqueuedMessages(t, r, tc.queue)
 		if len(enqueued) != 1 {
 			t.Errorf("%s;\nexpected queue %q to have one message; got %d messages in the queue.",
@@ -523,7 +648,7 @@ func TestEnqueueUnique(t *testing.T) {
 		h.FlushDB(t, r) // clean up db before each test case.
 
 		// Enqueue the task first. It should succeed.
-		err := c.Enqueue(tc.task, Unique(tc.ttl))
+		_, err := c.Enqueue(tc.task, Unique(tc.ttl))
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -535,7 +660,7 @@ func TestEnqueueUnique(t *testing.T) {
 		}
 
 		// Enqueue the task again. It should fail.
-		err = c.Enqueue(tc.task, Unique(tc.ttl))
+		_, err = c.Enqueue(tc.task, Unique(tc.ttl))
 		if err == nil {
 			t.Errorf("Enqueueing %+v did not return an error", tc.task)
 			continue
@@ -570,7 +695,7 @@ func TestEnqueueInUnique(t *testing.T) {
 		h.FlushDB(t, r) // clean up db before each test case.
 
 		// Enqueue the task first. It should succeed.
-		err := c.EnqueueIn(tc.d, tc.task, Unique(tc.ttl))
+		_, err := c.EnqueueIn(tc.d, tc.task, Unique(tc.ttl))
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -583,7 +708,7 @@ func TestEnqueueInUnique(t *testing.T) {
 		}
 
 		// Enqueue the task again. It should fail.
-		err = c.EnqueueIn(tc.d, tc.task, Unique(tc.ttl))
+		_, err = c.EnqueueIn(tc.d, tc.task, Unique(tc.ttl))
 		if err == nil {
 			t.Errorf("Enqueueing %+v did not return an error", tc.task)
 			continue
@@ -618,7 +743,7 @@ func TestEnqueueAtUnique(t *testing.T) {
 		h.FlushDB(t, r) // clean up db before each test case.
 
 		// Enqueue the task first. It should succeed.
-		err := c.EnqueueAt(tc.at, tc.task, Unique(tc.ttl))
+		_, err := c.EnqueueAt(tc.at, tc.task, Unique(tc.ttl))
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -631,7 +756,7 @@ func TestEnqueueAtUnique(t *testing.T) {
 		}
 
 		// Enqueue the task again. It should fail.
-		err = c.EnqueueAt(tc.at, tc.task, Unique(tc.ttl))
+		_, err = c.EnqueueAt(tc.at, tc.task, Unique(tc.ttl))
 		if err == nil {
 			t.Errorf("Enqueueing %+v did not return an error", tc.task)
 			continue
diff --git a/context_test.go b/context_test.go
index 1a5eff6..af11d1d 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/google/uuid"
 	"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:      uuid.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:      uuid.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: uuid.New(), Retry: 25, Retried: 0, Timeout: 1800}},
+		{"with non-zero retried message", &base.TaskMessage{Type: "something", ID: uuid.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..d811a46 100644
--- a/internal/asynqtest/asynqtest.go
+++ b/internal/asynqtest/asynqtest.go
@@ -13,8 +13,8 @@ import (
 	"github.com/go-redis/redis/v7"
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
+	"github.com/google/uuid"
 	"github.com/hibiken/asynq/internal/base"
-	"github.com/rs/xid"
 )
 
 // ZSetEntry is an entry in redis sorted set.
@@ -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:       uuid.New(),
+		Type:     taskType,
+		Queue:    base.DefaultQueueName,
+		Retry:    25,
+		Payload:  payload,
+		Timeout:  1800, // default timeout of 30 mins
+		Deadline: 0,    // no deadline
 	}
 }
 
@@ -87,7 +89,7 @@ func NewTaskMessage(taskType string, payload map[string]interface{}) *base.TaskM
 // task type, payload and queue name.
 func NewTaskMessageWithQueue(taskType string, payload map[string]interface{}, qname string) *base.TaskMessage {
 	return &base.TaskMessage{
-		ID:      xid.New(),
+		ID:      uuid.New(),
 		Type:    taskType,
 		Queue:   qname,
 		Retry:   25,
@@ -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..40f5c7a 100644
--- a/internal/base/base_test.go
+++ b/internal/base/base_test.go
@@ -12,7 +12,7 @@ import (
 	"time"
 
 	"github.com/google/go-cmp/cmp"
-	"github.com/rs/xid"
+	"github.com/google/uuid"
 )
 
 func TestQueueKey(t *testing.T) {
@@ -108,29 +108,31 @@ func TestWorkersKey(t *testing.T) {
 }
 
 func TestMessageEncoding(t *testing.T) {
-	id := xid.New()
+	id := uuid.New()
 	tests := []struct {
 		in  *TaskMessage
 		out *TaskMessage
 	}{
 		{
 			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/inspect_test.go b/internal/rdb/inspect_test.go
index 530df64..1b451b1 100644
--- a/internal/rdb/inspect_test.go
+++ b/internal/rdb/inspect_test.go
@@ -12,9 +12,9 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
+	"github.com/google/uuid"
 	h "github.com/hibiken/asynq/internal/asynqtest"
 	"github.com/hibiken/asynq/internal/base"
-	"github.com/rs/xid"
 )
 
 func TestCurrentStats(t *testing.T) {
@@ -618,7 +618,7 @@ func TestListScheduledPagination(t *testing.T) {
 func TestListRetry(t *testing.T) {
 	r := setup(t)
 	m1 := &base.TaskMessage{
-		ID:       xid.New(),
+		ID:       uuid.New(),
 		Type:     "send_email",
 		Queue:    "default",
 		Payload:  map[string]interface{}{"subject": "hello"},
@@ -627,7 +627,7 @@ func TestListRetry(t *testing.T) {
 		Retried:  10,
 	}
 	m2 := &base.TaskMessage{
-		ID:       xid.New(),
+		ID:       uuid.New(),
 		Type:     "reindex",
 		Queue:    "default",
 		Payload:  nil,
@@ -763,14 +763,14 @@ func TestListRetryPagination(t *testing.T) {
 func TestListDead(t *testing.T) {
 	r := setup(t)
 	m1 := &base.TaskMessage{
-		ID:       xid.New(),
+		ID:       uuid.New(),
 		Type:     "send_email",
 		Queue:    "default",
 		Payload:  map[string]interface{}{"subject": "hello"},
 		ErrorMsg: "email server not responding",
 	}
 	m2 := &base.TaskMessage{
-		ID:       xid.New(),
+		ID:       uuid.New(),
 		Type:     "reindex",
 		Queue:    "default",
 		Payload:  nil,
@@ -907,7 +907,7 @@ func TestEnqueueDeadTask(t *testing.T) {
 	tests := []struct {
 		dead         []h.ZSetEntry
 		score        int64
-		id           xid.ID
+		id           uuid.UUID
 		want         error // expected return value from calling EnqueueDeadTask
 		wantDead     []*base.TaskMessage
 		wantEnqueued map[string][]*base.TaskMessage
@@ -991,7 +991,7 @@ func TestEnqueueRetryTask(t *testing.T) {
 	tests := []struct {
 		retry        []h.ZSetEntry
 		score        int64
-		id           xid.ID
+		id           uuid.UUID
 		want         error // expected return value from calling EnqueueRetryTask
 		wantRetry    []*base.TaskMessage
 		wantEnqueued map[string][]*base.TaskMessage
@@ -1075,7 +1075,7 @@ func TestEnqueueScheduledTask(t *testing.T) {
 	tests := []struct {
 		scheduled     []h.ZSetEntry
 		score         int64
-		id            xid.ID
+		id            uuid.UUID
 		want          error // expected return value from calling EnqueueScheduledTask
 		wantScheduled []*base.TaskMessage
 		wantEnqueued  map[string][]*base.TaskMessage
@@ -1394,7 +1394,7 @@ func TestKillRetryTask(t *testing.T) {
 	tests := []struct {
 		retry     []h.ZSetEntry
 		dead      []h.ZSetEntry
-		id        xid.ID
+		id        uuid.UUID
 		score     int64
 		want      error
 		wantRetry []h.ZSetEntry
@@ -1471,7 +1471,7 @@ func TestKillScheduledTask(t *testing.T) {
 	tests := []struct {
 		scheduled     []h.ZSetEntry
 		dead          []h.ZSetEntry
-		id            xid.ID
+		id            uuid.UUID
 		score         int64
 		want          error
 		wantScheduled []h.ZSetEntry
@@ -1711,7 +1711,7 @@ func TestDeleteDeadTask(t *testing.T) {
 
 	tests := []struct {
 		dead     []h.ZSetEntry
-		id       xid.ID
+		id       uuid.UUID
 		score    int64
 		want     error
 		wantDead []*base.TaskMessage
@@ -1771,7 +1771,7 @@ func TestDeleteRetryTask(t *testing.T) {
 
 	tests := []struct {
 		retry     []h.ZSetEntry
-		id        xid.ID
+		id        uuid.UUID
 		score     int64
 		want      error
 		wantRetry []*base.TaskMessage
@@ -1823,7 +1823,7 @@ func TestDeleteScheduledTask(t *testing.T) {
 
 	tests := []struct {
 		scheduled     []h.ZSetEntry
-		id            xid.ID
+		id            uuid.UUID
 		score         int64
 		want          error
 		wantScheduled []*base.TaskMessage
diff --git a/internal/rdb/rdb_test.go b/internal/rdb/rdb_test.go
index ca3389d..805748f 100644
--- a/internal/rdb/rdb_test.go
+++ b/internal/rdb/rdb_test.go
@@ -14,9 +14,9 @@ import (
 	"github.com/go-redis/redis/v7"
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
+	"github.com/google/uuid"
 	h "github.com/hibiken/asynq/internal/asynqtest"
 	"github.com/hibiken/asynq/internal/base"
-	"github.com/rs/xid"
 )
 
 // TODO(hibiken): Get Redis address and db number from ENV variables.
@@ -73,7 +73,7 @@ func TestEnqueue(t *testing.T) {
 func TestEnqueueUnique(t *testing.T) {
 	r := setup(t)
 	m1 := base.TaskMessage{
-		ID:        xid.New(),
+		ID:        uuid.New(),
 		Type:      "email",
 		Payload:   map[string]interface{}{"user_id": 123},
 		Queue:     base.DefaultQueueName,
@@ -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:       uuid.New(),
+		Type:     "send_email",
+		Payload:  map[string]interface{}{"subject": "hello!"},
+		Timeout:  1800,
+		Deadline: 0,
+	}
+	t1Deadline := now.Unix() + t1.Timeout
+	t2 := &base.TaskMessage{
+		ID:       uuid.New(),
+		Type:     "export_csv",
+		Payload:  nil,
+		Timeout:  0,
+		Deadline: 1593021600,
+	}
+	t2Deadline := t2.Deadline
+	t3 := &base.TaskMessage{
+		ID:       uuid.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:       uuid.New(),
+		Type:     "send_email",
+		Payload:  nil,
+		Timeout:  1800,
+		Deadline: 0,
+	}
+	t1Deadline := now.Unix() + t1.Timeout
+	t2 := &base.TaskMessage{
+		ID:       uuid.New(),
+		Type:     "export_csv",
+		Payload:  nil,
+		Timeout:  0,
+		Deadline: 1592485787,
+	}
+	t2Deadline := t2.Deadline
 	t3 := &base.TaskMessage{
-		ID:        xid.New(),
+		ID:        uuid.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:      uuid.New(),
+		Type:    "send_email",
+		Payload: nil,
+		Queue:   "default",
+		Timeout: 1800,
+	}
+	t2 := &base.TaskMessage{
+		ID:      uuid.New(),
+		Type:    "export_csv",
+		Payload: nil,
+		Queue:   "default",
+		Timeout: 3000,
+	}
+	t3 := &base.TaskMessage{
+		ID:      uuid.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)
+		}
 	}
 }
 
@@ -506,7 +688,7 @@ func TestSchedule(t *testing.T) {
 func TestScheduleUnique(t *testing.T) {
 	r := setup(t)
 	m1 := base.TaskMessage{
-		ID:        xid.New(),
+		ID:        uuid.New(),
 		Type:      "email",
 		Payload:   map[string]interface{}{"user_id": 123},
 		Queue:     base.DefaultQueueName,
@@ -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:      uuid.New(),
+		Type:    "send_email",
+		Payload: map[string]interface{}{"subject": "Hola!"},
+		Retried: 10,
+		Timeout: 1800,
+	}
+	t1Deadline := now.Unix() + t1.Timeout
+	t2 := &base.TaskMessage{
+		ID:      uuid.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:      uuid.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:      uuid.New(),
+		Type:    "send_email",
+		Payload: nil,
+		Queue:   "default",
+		Retry:   25,
+		Retried: 0,
+		Timeout: 1800,
+	}
+	t1Deadline := now.Unix() + t1.Timeout
+	t2 := &base.TaskMessage{
+		ID:      uuid.New(),
+		Type:    "reindex",
+		Payload: nil,
+		Queue:   "default",
+		Retry:   25,
+		Retried: 0,
+		Timeout: 3000,
+	}
+	t2Deadline := now.Unix() + t2.Timeout
+	t3 := &base.TaskMessage{
+		ID:      uuid.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..2d7aa2a 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,
 		},
 	}
@@ -270,7 +257,7 @@ func TestProcessorRetry(t *testing.T) {
 			mu sync.Mutex // guards n
 			n  int        // number of times error handler is called
 		)
-		errHandler := func(t *Task, err error, retried, maxRetry int) {
+		errHandler := func(ctx context.Context, t *Task, err error) {
 			mu.Lock()
 			defer mu.Unlock()
 			n++
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/server_test.go b/server_test.go
index 82d1c72..4d08f45 100644
--- a/server_test.go
+++ b/server_test.go
@@ -41,12 +41,12 @@ func TestServer(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	err = c.Enqueue(NewTask("send_email", map[string]interface{}{"recipient_id": 123}))
+	_, err = c.Enqueue(NewTask("send_email", map[string]interface{}{"recipient_id": 123}))
 	if err != nil {
 		t.Errorf("could not enqueue a task: %v", err)
 	}
 
-	err = c.EnqueueAt(time.Now().Add(time.Hour), NewTask("send_email", map[string]interface{}{"recipient_id": 456}))
+	_, err = c.EnqueueAt(time.Now().Add(time.Hour), NewTask("send_email", map[string]interface{}{"recipient_id": 456}))
 	if err != nil {
 		t.Errorf("could not enqueue a task: %v", err)
 	}
@@ -183,15 +183,15 @@ func TestServerWithFlakyBroker(t *testing.T) {
 	}
 
 	for i := 0; i < 10; i++ {
-		err := c.Enqueue(NewTask("enqueued", nil), MaxRetry(i))
+		_, err := c.Enqueue(NewTask("enqueued", nil), MaxRetry(i))
 		if err != nil {
 			t.Fatal(err)
 		}
-		err = c.Enqueue(NewTask("bad_task", nil))
+		_, err = c.Enqueue(NewTask("bad_task", nil))
 		if err != nil {
 			t.Fatal(err)
 		}
-		err = c.EnqueueIn(time.Duration(i)*time.Second, NewTask("scheduled", nil))
+		_, err = c.EnqueueIn(time.Duration(i)*time.Second, NewTask("scheduled", nil))
 		if err != nil {
 			t.Fatal(err)
 		}
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()
+}