Add UniqueQueue interface and functions to create them

Author: zeripath

docs/content/doc/advanced/config-cheat-sheet.en-us.md

@@ -252,6 +252,10 @@ relation to port exhaustion.
 - `BATCH_LENGTH`: **20**: Batch data before passing to the handler
 - `CONN_STR`: **addrs=127.0.0.1:6379 db=0**: Connection string for the redis queue type.
 - `QUEUE_NAME`: **_queue**: The suffix for default redis queue name. Individual queues will default to **`name`**`QUEUE_NAME` but can be overriden in the specific `queue.name` section.
+- `SET_NAME`: **_unique**: The suffix that will added to the default redis
+set name for unique queues. Individual queues will default to
+**`name`**`QUEUE_NAME`_`SET_NAME`_ but can be overridden in the specific
+`queue.name` section.
 - `WRAP_IF_NECESSARY`: **true**: Will wrap queues with a timeoutable queue if the selected queue is not ready to be created - (Only relevant for the level queue.)
 - `MAX_ATTEMPTS`: **10**: Maximum number of attempts to create the wrapped queue
 - `TIMEOUT`: **GRACEFUL_HAMMER_TIME + 30s**: Timeout the creation of the wrapped queue if it takes longer than this to create.

modules/queue/queue.go

@@ -81,6 +81,16 @@ func (b *DummyQueue) Push(Data) error {
 	return nil
 }
 
+// PushFunc fakes a push of data to the queue with a function. The function is never run.
+func (b *DummyQueue) PushFunc(Data, func() error) error {
+	return nil
+}
+
+// Has always returns false as this queue never does anything
+func (b *DummyQueue) Has(Data) (bool, error) {
+	return false, nil
+}
+
 // Flush always returns nil
 func (b *DummyQueue) Flush(time.Duration) error {
 	return nil

modules/queue/setting.go

@@ -7,6 +7,7 @@ package queue
 import (
 	"encoding/json"
 	"fmt"
+	"strings"
 
 	"code.gitea.io/gitea/modules/log"
 	"code.gitea.io/gitea/modules/setting"
@@ -36,6 +37,7 @@ func getQueueSettings(name string) (setting.QueueSettings, []byte) {
 	opts["Password"] = q.Password
 	opts["DBIndex"] = q.DBIndex
 	opts["QueueName"] = q.QueueName
+	opts["SetName"] = q.SetName
 	opts["Workers"] = q.Workers
 	opts["MaxWorkers"] = q.MaxWorkers
 	opts["BlockTimeout"] = q.BlockTimeout
@@ -81,3 +83,38 @@ func CreateQueue(name string, handle HandlerFunc, exemplar interface{}) Queue {
 	}
 	return returnable
 }
+
+// CreateUniqueQueue for name with provided handler and exemplar
+func CreateUniqueQueue(name string, handle HandlerFunc, exemplar interface{}) UniqueQueue {
+	q, cfg := getQueueSettings(name)
+	if len(cfg) == 0 {
+		return nil
+	}
+
+	if len(q.Type) > 0 && q.Type != "dummy" && !strings.HasPrefix(q.Type, "unique-") {
+		q.Type = "unique-" + q.Type
+	}
+
+	typ, err := validType(q.Type)
+	if err != nil || typ == PersistableChannelQueueType {
+		typ = PersistableChannelUniqueQueueType
+	}
+
+	returnable, err := NewQueue(typ, handle, cfg, exemplar)
+	if q.WrapIfNecessary && err != nil {
+		log.Warn("Unable to create unique queue for %s: %v", name, err)
+		log.Warn("Attempting to create wrapped queue")
+		returnable, err = NewQueue(WrappedUniqueQueueType, handle, WrappedUniqueQueueConfiguration{
+			Underlying:  typ,
+			Timeout:     q.Timeout,
+			MaxAttempts: q.MaxAttempts,
+			Config:      cfg,
+			QueueLength: q.Length,
+		}, exemplar)
+	}
+	if err != nil {
+		log.Error("Unable to create unique queue for %s: %v", name, err)
+		return nil
+	}
+	return returnable.(UniqueQueue)
+}

modules/queue/unique_queue.go

@@ -0,0 +1,29 @@
+// Copyright 2020 The Gitea Authors. All rights reserved.
+// Use of this source code is governed by a MIT-style
+// license that can be found in the LICENSE file.
+
+package queue
+
+import (
+	"context"
+	"fmt"
+)
+
+// UniqueQueue defines a queue which guarantees only one instance of same
+// data is in the queue. Instances with same identity will be
+// discarded if there is already one in the line.
+//
+// This queue is particularly useful for preventing duplicated task
+// of same purpose.
+//
+// Users of this queue should be careful to push only the identifier of the
+// data
+type UniqueQueue interface {
+	Run(atShutdown, atTerminate func(context.Context, func()))
+	Push(Data) error
+	PushFunc(Data, func() error) error
+	Has(Data) (bool, error)
+}
+
+// ErrAlreadyInQueue is returned when trying to push data to the queue that is already in the queue
+var ErrAlreadyInQueue = fmt.Errorf("already in queue")

modules/setting/queue.go

@@ -26,6 +26,7 @@ type QueueSettings struct {
 	Addresses        string
 	Password         string
 	QueueName        string
+	SetName          string
 	DBIndex          int
 	WrapIfNecessary  bool
 	MaxAttempts      int
@@ -54,8 +55,13 @@ func GetQueueSettings(name string) QueueSettings {
 			q.DataDir = key.MustString(q.DataDir)
 		case "QUEUE_NAME":
 			q.QueueName = key.MustString(q.QueueName)
+		case "SET_NAME":
+			q.SetName = key.MustString(q.SetName)
 		}
 	}
+	if len(q.SetName) == 0 && len(Queue.SetName) > 0 {
+		q.SetName = q.QueueName + Queue.SetName
+	}
 	if !filepath.IsAbs(q.DataDir) {
 		q.DataDir = filepath.Join(AppDataPath, q.DataDir)
 	}
@@ -100,6 +106,7 @@ func NewQueueService() {
 	Queue.BoostTimeout = sec.Key("BOOST_TIMEOUT").MustDuration(5 * time.Minute)
 	Queue.BoostWorkers = sec.Key("BOOST_WORKERS").MustInt(5)
 	Queue.QueueName = sec.Key("QUEUE_NAME").MustString("_queue")
+	Queue.SetName = sec.Key("SET_NAME").MustString("")
 
 	// Now handle the old issue_indexer configuration
 	section := Cfg.Section("queue.issue_indexer")