v0.17.2

More details: https://redis.io/commands/KEYS

CHANGELOG.md

@@ -7,6 +7,30 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.17.2] - 2021-06-06
+
+### Fixed
+
+- Free unique lock when task is deleted (https://github.com/hibiken/asynq/issues/275).
+
+## [0.17.1] - 2021-04-04
+
+### Fixed
+
+- Fix bug in internal `RDB.memoryUsage` method.
+
+## [0.17.0] - 2021-03-24
+
+### Added
+
+- `DialTimeout`, `ReadTimeout`, and `WriteTimeout` options are added to `RedisConnOpt`.
+
+## [0.16.1] - 2021-03-20
+
+### Fixed
+
+- Replace `KEYS` command with `SCAN` as recommended by [redis doc](https://redis.io/commands/KEYS).
+
 ## [0.16.0] - 2021-03-10
 
 ### Added

README.md

@@ -1,14 +1,14 @@
-# Asynq
+<img src="https://user-images.githubusercontent.com/11155743/114697792-ffbfa580-9d26-11eb-8e5b-33bef69476dc.png" alt="Asynq logo" width="360px" />
+
+# Simple, reliable & efficient distributed task queue in Go
 
-![Build Status](https://github.com/hibiken/asynq/workflows/build/badge.svg)
 [![GoDoc](https://godoc.org/github.com/hibiken/asynq?status.svg)](https://godoc.org/github.com/hibiken/asynq)
 [![Go Report Card](https://goreportcard.com/badge/github.com/hibiken/asynq)](https://goreportcard.com/report/github.com/hibiken/asynq)
+![Build Status](https://github.com/hibiken/asynq/workflows/build/badge.svg)
 [![License: MIT](https://img.shields.io/badge/license-MIT-green.svg)](https://opensource.org/licenses/MIT)
 [![Gitter chat](https://badges.gitter.im/go-asynq/gitter.svg)](https://gitter.im/go-asynq/community)
 
-## Overview
-
-Asynq is a Go library for queueing tasks and processing them asynchronously with workers. It's backed by Redis and is designed to be scalable yet easy to get started.
+Asynq is a Go library for queueing tasks and processing them asynchronously with workers. It's backed by [Redis](https://redis.io/) and is designed to be scalable yet easy to get started.
 
 Highlevel overview of how Asynq works:
 
@@ -16,16 +16,11 @@ Highlevel overview of how Asynq works:
 - Server pulls task off queues and starts a worker goroutine for each task
 - Tasks are processed concurrently by multiple workers
 
-Task queues are used as a mechanism to distribute work across multiple machines.  
-A system can consist of multiple worker servers and brokers, giving way to high availability and horizontal scaling.
+Task queues are used as a mechanism to distribute work across multiple machines. A system can consist of multiple worker servers and brokers, giving way to high availability and horizontal scaling.
 
-![Task Queue Diagram](/docs/assets/overview.png)
+**Example use case**
 
-## Stability and Compatibility
-
-**Important Note**: Current major version is zero (v0.x.x) to accomodate rapid development and fast iteration while getting early feedback from users (Feedback on APIs are appreciated!). The public API could change without a major version update before v1.0.0 release.
-
-**Status**: The library is currently undergoing heavy development with frequent, breaking API changes.
+![Task Queue Diagram](https://user-images.githubusercontent.com/11155743/116358505-656f5f80-a806-11eb-9c16-94e49dab0f99.jpg)
 
 ## Features
 
@@ -47,14 +42,24 @@ A system can consist of multiple worker servers and brokers, giving way to high
 - [Web UI](#web-ui) to inspect and remote-control queues and tasks
 - [CLI](#command-line-tool) to inspect and remote-control queues and tasks
 
+## Stability and Compatibility
+
+**Status**: The library is currently undergoing **heavy development** with frequent, breaking API changes.
+
+> ☝️ **Important Note**: Current major version is zero (`v0.x.x`) to accomodate rapid development and fast iteration while getting early feedback from users (_feedback on APIs are appreciated!_). The public API could change without a major version update before `v1.0.0` release.
+
 ## Quickstart
 
-First, make sure you are running a Redis server locally.
+Make sure you have Go installed ([download](https://golang.org/dl/)). Version `1.13` or higher is required. 
+
+Initialize your project by creating a folder and then running `go mod init github.com/your/repo` ([learn more](https://blog.golang.org/using-go-modules)) inside the folder. Then install Asynq library with the [`go get`](https://golang.org/cmd/go/#hdr-Add_dependencies_to_current_module_and_install_them) command:
 
 ```sh
-$ redis-server
+go get -u github.com/hibiken/asynq
 ```
 
+Make sure you're running a Redis server locally or from a [Docker](https://hub.docker.com/_/redis) container. Version `3.0` or higher is required. 
+
 Next, write a package that encapsulates task creation and task handling.
 
 ```go
@@ -204,10 +209,9 @@ func main() {
 }
 ```
 
-Next, start a worker server to process these tasks in the background.  
-To start the background workers, use [`Server`](https://pkg.go.dev/github.com/hibiken/asynq?tab=doc#Server) and provide your [`Handler`](https://pkg.go.dev/github.com/hibiken/asynq?tab=doc#Handler) to process the tasks.
+Next, start a worker server to process these tasks in the background. To start the background workers, use [`Server`](https://pkg.go.dev/github.com/hibiken/asynq?tab=doc#Server) and provide your [`Handler`](https://pkg.go.dev/github.com/hibiken/asynq?tab=doc#Handler) to process the tasks.
 
-You can optionally use [`ServeMux`](https://pkg.go.dev/github.com/hibiken/asynq?tab=doc#ServeMux) to create a handler, just as you would with [`"net/http"`](https://golang.org/pkg/net/http/) Handler.
+You can optionally use [`ServeMux`](https://pkg.go.dev/github.com/hibiken/asynq?tab=doc#ServeMux) to create a handler, just as you would with [`net/http`](https://golang.org/pkg/net/http/) Handler.
 
 ```go
 package main
@@ -248,65 +252,52 @@ func main() {
 }
 ```
 
-For a more detailed walk-through of the library, see our [Getting Started Guide](https://github.com/hibiken/asynq/wiki/Getting-Started).
+For a more detailed walk-through of the library, see our [Getting Started](https://github.com/hibiken/asynq/wiki/Getting-Started) guide.
 
-To Learn more about `asynq` features and APIs, see our [Wiki](https://github.com/hibiken/asynq/wiki) and [godoc](https://godoc.org/github.com/hibiken/asynq).
+To learn more about `asynq` features and APIs, see the package [godoc](https://godoc.org/github.com/hibiken/asynq).
 
 ## Web UI
 
 [Asynqmon](https://github.com/hibiken/asynqmon) is a web based tool for monitoring and administrating Asynq queues and tasks.
-Please see the tool's [README](https://github.com/hibiken/asynqmon) for details.
 
-Here's a few screenshots of the web UI.
+Here's a few screenshots of the Web UI:
 
 **Queues view**  
-![Web UI QueuesView](/docs/assets/asynqmon-queues-view.png)
+
+![Web UI Queues View](https://user-images.githubusercontent.com/11155743/114697016-07327f00-9d26-11eb-808c-0ac841dc888e.png)
 
 **Tasks view**  
-![Web UI TasksView](/docs/assets/asynqmon-task-view.png)
+
+![Web UI TasksView](https://user-images.githubusercontent.com/11155743/114697070-1f0a0300-9d26-11eb-855c-d3ec263865b7.png)
+
+**Settings and adaptive dark mode**
+
+![Web UI Settings and adaptive dark mode](https://user-images.githubusercontent.com/11155743/114697149-3517c380-9d26-11eb-9f7a-ae2dd00aad5b.png)
+
+For details on how to use the tool, refer to the tool's [README](https://github.com/hibiken/asynqmon#readme). 
 
 ## Command Line Tool
 
 Asynq ships with a command line tool to inspect the state of queues and tasks.
 
-Here's an example of running the `stats` command.
-
-![Gif](/docs/assets/demo.gif)
-
-For details on how to use the tool, refer to the tool's [README](/tools/asynq/README.md).
-
-## Installation
-
-To install `asynq` library, run the following command:
-
-```sh
-go get -u github.com/hibiken/asynq
-```
-
 To install the CLI tool, run the following command:
 
 ```sh
 go get -u github.com/hibiken/asynq/tools/asynq
 ```
 
-## Requirements
+Here's an example of running the `asynq stats` command:
 
-| Dependency                 | Version |
-| -------------------------- | ------- |
-| [Redis](https://redis.io/) | v3.0+   |
-| [Go](https://golang.org/)  | v1.13+  |
+![Gif](/docs/assets/demo.gif)
+
+For details on how to use the tool, refer to the tool's [README](/tools/asynq/README.md).
 
 ## Contributing
 
-We are open to, and grateful for, any contributions (Github issues/pull-requests, feedback on Gitter channel, etc) made by the community.
+We are open to, and grateful for, any contributions (GitHub issues/PRs, feedback on [Gitter channel](https://gitter.im/go-asynq/community), etc) made by the community.
+
 Please see the [Contribution Guide](/CONTRIBUTING.md) before contributing.
 
-## Acknowledgements
-
-- [Sidekiq](https://github.com/mperham/sidekiq) : Many of the design ideas are taken from sidekiq and its Web UI
-- [RQ](https://github.com/rq/rq) : Client APIs are inspired by rq library.
-- [Cobra](https://github.com/spf13/cobra) : Asynq CLI is built with cobra
-
 ## License
 
-Asynq is released under the MIT license. See [LICENSE](https://github.com/hibiken/asynq/blob/master/LICENSE).
+Copyright (c) 2019-present [Ken Hibino](https://github.com/hibiken) and [Contributors](https://github.com/hibiken/asynq/graphs/contributors). `Asynq` is free and open-source software licensed under the [MIT License](https://github.com/hibiken/asynq/blob/master/LICENSE). Official logo was created by [Vic Shóstak](https://github.com/koddr) and distributed under [Creative Commons](https://creativecommons.org/publicdomain/zero/1.0/) license (CC0 1.0 Universal).

asynq.go

@@ -10,6 +10,7 @@ import (
 	"net/url"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/go-redis/redis/v7"
 )
@@ -68,6 +69,26 @@ type RedisClientOpt struct {
 	// See: https://redis.io/commands/select.
 	DB int
 
+	// Dial timeout for establishing new connections.
+	// Default is 5 seconds.
+	DialTimeout time.Duration
+
+	// Timeout for socket reads.
+	// If timeout is reached, read commands will fail with a timeout error
+	// instead of blocking.
+	//
+	// Use value -1 for no timeout and 0 for default.
+	// Default is 3 seconds.
+	ReadTimeout time.Duration
+
+	// Timeout for socket writes.
+	// If timeout is reached, write commands will fail with a timeout error
+	// instead of blocking.
+	//
+	// Use value -1 for no timeout and 0 for default.
+	// Default is ReadTimout.
+	WriteTimeout time.Duration
+
 	// Maximum number of socket connections.
 	// Default is 10 connections per every CPU as reported by runtime.NumCPU.
 	PoolSize int
@@ -79,13 +100,16 @@ type RedisClientOpt struct {
 
 func (opt RedisClientOpt) MakeRedisClient() interface{} {
 	return redis.NewClient(&redis.Options{
-		Network:   opt.Network,
-		Addr:      opt.Addr,
-		Username:  opt.Username,
-		Password:  opt.Password,
-		DB:        opt.DB,
-		PoolSize:  opt.PoolSize,
-		TLSConfig: opt.TLSConfig,
+		Network:      opt.Network,
+		Addr:         opt.Addr,
+		Username:     opt.Username,
+		Password:     opt.Password,
+		DB:           opt.DB,
+		DialTimeout:  opt.DialTimeout,
+		ReadTimeout:  opt.ReadTimeout,
+		WriteTimeout: opt.WriteTimeout,
+		PoolSize:     opt.PoolSize,
+		TLSConfig:    opt.TLSConfig,
 	})
 }
 
@@ -116,6 +140,26 @@ type RedisFailoverClientOpt struct {
 	// See: https://redis.io/commands/select.
 	DB int
 
+	// Dial timeout for establishing new connections.
+	// Default is 5 seconds.
+	DialTimeout time.Duration
+
+	// Timeout for socket reads.
+	// If timeout is reached, read commands will fail with a timeout error
+	// instead of blocking.
+	//
+	// Use value -1 for no timeout and 0 for default.
+	// Default is 3 seconds.
+	ReadTimeout time.Duration
+
+	// Timeout for socket writes.
+	// If timeout is reached, write commands will fail with a timeout error
+	// instead of blocking.
+	//
+	// Use value -1 for no timeout and 0 for default.
+	// Default is ReadTimeout
+	WriteTimeout time.Duration
+
 	// Maximum number of socket connections.
 	// Default is 10 connections per every CPU as reported by runtime.NumCPU.
 	PoolSize int
@@ -133,6 +177,9 @@ func (opt RedisFailoverClientOpt) MakeRedisClient() interface{} {
 		Username:         opt.Username,
 		Password:         opt.Password,
 		DB:               opt.DB,
+		DialTimeout:      opt.DialTimeout,
+		ReadTimeout:      opt.ReadTimeout,
+		WriteTimeout:     opt.WriteTimeout,
 		PoolSize:         opt.PoolSize,
 		TLSConfig:        opt.TLSConfig,
 	})
@@ -157,6 +204,26 @@ type RedisClusterClientOpt struct {
 	// See: https://redis.io/commands/auth.
 	Password string
 
+	// Dial timeout for establishing new connections.
+	// Default is 5 seconds.
+	DialTimeout time.Duration
+
+	// Timeout for socket reads.
+	// If timeout is reached, read commands will fail with a timeout error
+	// instead of blocking.
+	//
+	// Use value -1 for no timeout and 0 for default.
+	// Default is 3 seconds.
+	ReadTimeout time.Duration
+
+	// Timeout for socket writes.
+	// If timeout is reached, write commands will fail with a timeout error
+	// instead of blocking.
+	//
+	// Use value -1 for no timeout and 0 for default.
+	// Default is ReadTimeout.
+	WriteTimeout time.Duration
+
 	// TLS Config used to connect to a server.
 	// TLS will be negotiated only if this field is set.
 	TLSConfig *tls.Config
@@ -168,6 +235,9 @@ func (opt RedisClusterClientOpt) MakeRedisClient() interface{} {
 		MaxRedirects: opt.MaxRedirects,
 		Username:     opt.Username,
 		Password:     opt.Password,
+		DialTimeout:  opt.DialTimeout,
+		ReadTimeout:  opt.ReadTimeout,
+		WriteTimeout: opt.WriteTimeout,
 		TLSConfig:    opt.TLSConfig,
 	})
 }

internal/base/base.go

@@ -19,7 +19,7 @@ import (
 )
 
 // Version of asynq library and CLI.
-const Version = "0.16.0"
+const Version = "0.17.2"
 
 // DefaultQueueName is the queue name used if none are specified by user.
 const DefaultQueueName = "default"

internal/rdb/inspect.go

@@ -172,9 +172,21 @@ func (r *RDB) CurrentStats(qname string) (*Stats, error) {
 }
 
 func (r *RDB) memoryUsage(qname string) (int64, error) {
-	keys, err := r.client.Keys(fmt.Sprintf("asynq:{%s}*", qname)).Result()
-	if err != nil {
-		return 0, err
+	var (
+		keys   []string
+		data   []string
+		cursor uint64
+		err    error
+	)
+	for {
+		data, cursor, err = r.client.Scan(cursor, fmt.Sprintf("asynq:{%s}*", qname), 100).Result()
+		if err != nil {
+			return 0, err
+		}
+		keys = append(keys, data...)
+		if cursor == 0 {
+			break
+		}
 	}
 	var usg int64
 	for _, k := range keys {
@@ -723,6 +735,11 @@ func (r *RDB) DeletePendingTask(qname string, id uuid.UUID) error {
 			if n == 0 {
 				return ErrTaskNotFound
 			}
+			if r.client.Get(msg.UniqueKey).Val() == msg.ID.String() {
+				if err := r.client.Del(msg.UniqueKey).Err(); err != nil {
+					return err
+				}
+			}
 			return nil
 		}
 	}
@@ -735,6 +752,9 @@ for _, msg in ipairs(msgs) do
 	local decoded = cjson.decode(msg)
 	if decoded["ID"] == ARGV[2] then
 		redis.call("ZREM", KEYS[1], msg)
+		if redis.call("GET", decoded["UniqueKey"]) == ARGV[2] then
+			redis.call("DEL", decoded["UniqueKey"])
+		end
 		return 1
 	end
 end
@@ -757,9 +777,15 @@ func (r *RDB) deleteTask(key, id string, score float64) error {
 
 // KEYS[1] -> queue to delete
 var deleteAllCmd = redis.NewScript(`
-local n = redis.call("ZCARD", KEYS[1])
+local msgs = redis.call("ZRANGE", KEYS[1], 0, -1)
+for _, msg in ipairs(msgs) do
+	local decoded = cjson.decode(msg)
+	if redis.call("GET", decoded["UniqueKey"]) == decoded["ID"] then
+		redis.call("DEL", decoded["UniqueKey"])
+	end
+end
 redis.call("DEL", KEYS[1])
-return n`)
+return table.getn(msgs)`)
 
 // DeleteAllArchivedTasks deletes all archived tasks from the given queue
 // and returns the number of tasks deleted.
@@ -793,9 +819,15 @@ func (r *RDB) deleteAll(key string) (int64, error) {
 
 // KEYS[1] -> asynq:{<qname>}
 var deleteAllPendingCmd = redis.NewScript(`
-local n = redis.call("LLEN", KEYS[1])
+local msgs = redis.call("LRANGE", KEYS[1], 0, -1)
+for _, msg in ipairs(msgs) do
+	local decoded = cjson.decode(msg)
+	if redis.call("GET", decoded["UniqueKey"]) == decoded["ID"] then
+		redis.call("DEL", decoded["UniqueKey"])
+	end
+end
 redis.call("DEL", KEYS[1])
-return n`)
+return table.getn(msgs)`)
 
 // DeleteAllPendingTasks deletes all pending tasks from the given queue
 // and returns the number of tasks deleted.

internal/rdb/inspect_test.go

@@ -2445,7 +2445,68 @@ func TestDeleteScheduledTask(t *testing.T) {
 	}
 }
 
-func TestDeleteAllDeadTasks(t *testing.T) {
+func TestDeleteUniqueTask(t *testing.T) {
+	r := setup(t)
+	defer r.Close()
+	m1 := &base.TaskMessage{
+		ID:        uuid.New(),
+		Type:      "reindex",
+		Payload:   nil,
+		Timeout:   1800,
+		Deadline:  0,
+		UniqueKey: "asynq:{default}:unique:reindex:nil",
+		Queue:     "default",
+	}
+	t1 := time.Now().Add(5 * time.Minute)
+
+	tests := []struct {
+		scheduled     map[string][]base.Z
+		qname         string
+		id            uuid.UUID
+		score         int64
+		uniqueKey     string
+		wantScheduled map[string][]*base.TaskMessage
+	}{
+		{
+			scheduled: map[string][]base.Z{
+				"default": {
+					{Message: m1, Score: t1.Unix()},
+				},
+			},
+			qname:     "default",
+			id:        m1.ID,
+			score:     t1.Unix(),
+			uniqueKey: m1.UniqueKey,
+			wantScheduled: map[string][]*base.TaskMessage{
+				"default": {},
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		h.FlushDB(t, r.client) // clean up db before each test case
+		h.SeedAllScheduledQueues(t, r.client, tc.scheduled)
+		if err := r.client.SetNX(tc.uniqueKey, tc.id.String(), time.Minute).Err(); err != nil {
+			t.Fatalf("Could not set unique lock in redis: %v", err)
+		}
+
+		if err := r.DeleteScheduledTask(tc.qname, tc.id, tc.score); err != nil {
+			t.Errorf("r.DeleteScheduledTask(%q, %v, %v) returned error: %v", tc.qname, tc.id, tc.score, err)
+			continue
+		}
+
+		for qname, want := range tc.wantScheduled {
+			gotScheduled := h.GetScheduledMessages(t, r.client, qname)
+			if diff := cmp.Diff(want, gotScheduled, h.SortMsgOpt); diff != "" {
+				t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.ScheduledKey(qname), diff)
+			}
+		}
+		if r.client.Exists(tc.uniqueKey).Val() != 0 {
+			t.Errorf("Uniqueness lock %q still exists", tc.uniqueKey)
+		}
+	}
+}
+func TestDeleteAllArchivedTasks(t *testing.T) {
 	r := setup(t)
 	defer r.Close()
 	m1 := h.NewTaskMessage("task1", nil)
@@ -2507,6 +2568,89 @@ func TestDeleteAllDeadTasks(t *testing.T) {
 	}
 }
 
+func TestDeleteAllArchivedTasksWithUniqueKey(t *testing.T) {
+	r := setup(t)
+	defer r.Close()
+	m1 := &base.TaskMessage{
+		ID:        uuid.New(),
+		Type:      "task1",
+		Payload:   nil,
+		Timeout:   1800,
+		Deadline:  0,
+		UniqueKey: "asynq:{default}:unique:task1:nil",
+		Queue:     "default",
+	}
+	m2 := &base.TaskMessage{
+		ID:        uuid.New(),
+		Type:      "task2",
+		Payload:   nil,
+		Timeout:   1800,
+		Deadline:  0,
+		UniqueKey: "asynq:{default}:unique:task2:nil",
+		Queue:     "default",
+	}
+	m3 := h.NewTaskMessage("task3", nil)
+
+	tests := []struct {
+		archived     map[string][]base.Z
+		qname        string
+		want         int64
+		wantArchived map[string][]*base.TaskMessage
+	}{
+		{
+			archived: map[string][]base.Z{
+				"default": {
+					{Message: m1, Score: time.Now().Unix()},
+					{Message: m2, Score: time.Now().Unix()},
+					{Message: m3, Score: time.Now().Unix()},
+				},
+			},
+			qname: "default",
+			want:  3,
+			wantArchived: map[string][]*base.TaskMessage{
+				"default": {},
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		h.FlushDB(t, r.client) // clean up db before each test case
+		h.SeedAllArchivedQueues(t, r.client, tc.archived)
+		var uniqueKeys []string // list of unique keys set in redis
+		for _, zs := range tc.archived {
+			for _, z := range zs {
+				if len(z.Message.UniqueKey) > 0 {
+					err := r.client.SetNX(z.Message.UniqueKey, z.Message.ID.String(), time.Minute).Err()
+					if err != nil {
+						t.Fatalf("Failed to set unique lock in redis: %v", err)
+					}
+					uniqueKeys = append(uniqueKeys, z.Message.UniqueKey)
+				}
+			}
+		}
+
+		got, err := r.DeleteAllArchivedTasks(tc.qname)
+		if err != nil {
+			t.Errorf("r.DeleteAllDeadTasks(%q) returned error: %v", tc.qname, err)
+		}
+		if got != tc.want {
+			t.Errorf("r.DeleteAllDeadTasks(%q) = %d, nil, want %d, nil", tc.qname, got, tc.want)
+		}
+		for qname, want := range tc.wantArchived {
+			gotArchived := h.GetArchivedMessages(t, r.client, qname)
+			if diff := cmp.Diff(want, gotArchived, h.SortMsgOpt); diff != "" {
+				t.Errorf("mismatch found in %q; (-want, +got)\n%s", base.ArchivedKey(qname), diff)
+			}
+		}
+
+		for _, uniqueKey := range uniqueKeys {
+			if r.client.Exists(uniqueKey).Val() != 0 {
+				t.Errorf("Uniqueness lock %q still exists", uniqueKey)
+			}
+		}
+	}
+}
+
 func TestDeleteAllRetryTasks(t *testing.T) {
 	r := setup(t)
 	defer r.Close()