Add doctor commands to interrogate levelDB queues

modules/doctor/queue.go

@@ -0,0 +1,210 @@
+// Copyright 2022 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package doctor
+
+import (
+	"context"
+
+	"code.gitea.io/gitea/modules/log"
+	"code.gitea.io/gitea/modules/nosql"
+	"code.gitea.io/gitea/modules/queue"
+	"code.gitea.io/gitea/modules/setting"
+
+	"gitea.com/lunny/levelqueue"
+)
+
+var levelqueueTypes = []string{
+	string(queue.PersistableChannelQueueType),
+	string(queue.PersistableChannelUniqueQueueType),
+	string(queue.LevelQueueType),
+	string(queue.LevelUniqueQueueType),
+}
+
+func checkUniqueQueues(ctx context.Context, logger log.Logger, autofix bool) error {
+	for _, name := range queue.KnownUniqueQueueNames {
+		q := setting.GetQueueSettings(string(name))
+		if q.Type == "" {
+			q.Type = string(queue.PersistableChannelQueueType)
+		}
+		found := false
+		for _, typ := range levelqueueTypes {
+			if typ == q.Type {
+				found = true
+				break
+			}
+		}
+		if !found {
+			logger.Info("Queue: %s\nType: %s\nNo LevelDB", q.Name, q.Type)
+			continue
+		}
+
+		connection := q.ConnectionString
+		if connection == "" {
+			connection = q.DataDir
+		}
+
+		db, err := nosql.GetManager().GetLevelDB(connection)
+		if err != nil {
+			logger.Error("Queue: %s\nUnable to open DB connection %q: %v", q.Name, connection, err)
+			return err
+		}
+		defer db.Close()
+
+		prefix := q.Name
+
+		iQueue, err := levelqueue.NewQueue(db, []byte(prefix), false)
+		if err != nil {
+			logger.Error("Queue: %s\nUnable to open Queue component: %v", q.Name, err)
+			return err
+		}
+
+		iSet, err := levelqueue.NewSet(db, []byte(prefix+"-unique"), false)
+		if err != nil {
+			logger.Error("Queue: %s\nUnable to open Set component: %v", q.Name, err)
+			return err
+		}
+
+		qLen := iQueue.Len()
+		sMembers, err := iSet.Members()
+		if err != nil {
+			logger.Error("Queue: %s\nUnable to get members of Set component: %v", q.Name, err)
+			return err
+		}
+		sLen := len(sMembers)
+
+		if int(qLen) == sLen {
+			if qLen == 0 {
+				logger.Info("Queue: %s\nType: %s\nLevelDB: %s", q.Name, q.Type, "empty")
+			} else {
+				logger.Info("Queue: %s\nType: %s\nLevelDB contains: %d entries", q.Name, q.Type, qLen)
+			}
+			continue
+		}
+		logger.Warn("Queue: %s\nType: %s\nContains different numbers of elements in Queue component %d to Set component %d", q.Name, q.Type, qLen, sLen)
+		if !autofix {
+			continue
+		}
+
+		// Empty out the old set members
+		for _, member := range sMembers {
+			_, err := iSet.Remove(member)
+			if err != nil {
+				logger.Error("Queue: %s\nUnable to remove Set member %s: %v", q.Name, string(member), err)
+				return err
+			}
+		}
+
+		// Now iterate across the queue
+		for i := int64(0); i < qLen; i++ {
+			// Pop from the left
+			qData, err := iQueue.LPop()
+			if err != nil {
+				logger.Error("Queue: %s\nUnable to LPop out: %v", q.Name, err)
+				return err
+			}
+			// And add to the right
+			err = iQueue.RPush(qData)
+			if err != nil {
+				logger.Error("Queue: %s\nUnable to RPush back: %v", q.Name, err)
+				return err
+			}
+			// And add back to the set
+			_, err = iSet.Add(qData)
+			if err != nil {
+				logger.Error("Queue: %s\nUnable to add back in to Set: %v", q.Name, err)
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func queueListDB(ctx context.Context, logger log.Logger, autofix bool) error {
+	connections := []string{}
+	queueNames := make([]string, 0, len(queue.KnownUniqueQueueNames)+len(queue.KnownQueueNames))
+	for _, name := range queue.KnownUniqueQueueNames {
+		queueNames = append(queueNames, string(name))
+	}
+	for _, name := range queue.KnownQueueNames {
+		queueNames = append(queueNames, string(name))
+	}
+
+	for _, name := range queueNames {
+		q := setting.GetQueueSettings(name)
+		if q.Type == "" {
+			q.Type = string(queue.PersistableChannelQueueType)
+		}
+		found := false
+		for _, typ := range levelqueueTypes {
+			if typ == q.Type {
+				found = true
+				break
+			}
+		}
+		if !found {
+			continue
+		}
+		if q.ConnectionString != "" {
+			found := false
+			for _, connection := range connections {
+				if connection == q.ConnectionString {
+					found = true
+					break
+				}
+			}
+			if !found {
+				connections = append(connections, q.ConnectionString)
+			}
+			continue
+		}
+		found = false
+		for _, connection := range connections {
+			if connection == q.DataDir {
+				found = true
+				break
+			}
+		}
+		if !found {
+			connections = append(connections, q.DataDir)
+		}
+	}
+
+	for _, connection := range connections {
+		logger.Info("LevelDB: %s", connection)
+		db, err := nosql.GetManager().GetLevelDB(connection)
+		if err != nil {
+			logger.Error("Connection: %q Unable to open DB: %v", connection, err)
+			return err
+		}
+		defer db.Close()
+		iter := db.NewIterator(nil, nil)
+		for iter.Next() {
+			logger.Info("%s\n%s", log.NewColoredIDValue(string(iter.Key())), string(iter.Value()))
+		}
+		iter.Release()
+	}
+	return nil
+}
+
+func init() {
+	Register(&Check{
+		Title:                      "Check if there are corrupt level uniquequeues",
+		Name:                       "uniquequeues-corrupt",
+		IsDefault:                  false,
+		Run:                        checkUniqueQueues,
+		AbortIfFailed:              false,
+		SkipDatabaseInitialization: false,
+		Priority:                   1,
+	})
+
+	Register(&Check{
+		Title:                      "List all entries in leveldb",
+		Name:                       "queues-listdb",
+		IsDefault:                  false,
+		Run:                        queueListDB,
+		AbortIfFailed:              false,
+		SkipDatabaseInitialization: false,
+		Priority:                   1,
+	})
+}

modules/indexer/code/indexer.go

@@ -165,7 +165,7 @@ func Init() {
 			return unhandled
 		}
 
-		indexerQueue = queue.CreateUniqueQueue("code_indexer", handler, &IndexerData{})
+		indexerQueue = queue.CreateUniqueQueue(queue.CodeIndexerQueueName, handler, &IndexerData{})
 		if indexerQueue == nil {
 			log.Fatal("Unable to create codes indexer queue")
 		}

modules/indexer/issues/indexer.go

@@ -157,7 +157,7 @@ func InitIssueIndexer(syncReindex bool) {
 			return nil
 		}
 
-		issueIndexerQueue = queue.CreateQueue("issue_indexer", handler, &IndexerData{})
+		issueIndexerQueue = queue.CreateQueue(queue.IssueIndexerQueueName, handler, &IndexerData{})
 
 		if issueIndexerQueue == nil {
 			log.Fatal("Unable to create issue indexer queue")

modules/indexer/stats/queue.go

@@ -27,9 +27,9 @@ func handle(data ...queue.Data) []queue.Data {
 }
 
 func initStatsQueue() error {
-	statsQueue = queue.CreateUniqueQueue("repo_stats_update", handle, int64(0))
+	statsQueue = queue.CreateUniqueQueue(queue.RepoStatsUpdateQueueName, handle, int64(0))
 	if statsQueue == nil {
-		return fmt.Errorf("Unable to create repo_stats_update Queue")
+		return fmt.Errorf("unable to create repo_stats_update Queue")
 	}
 
 	go graceful.GetManager().RunWithShutdownFns(statsQueue.Run)

modules/mirror/mirror.go

@@ -33,7 +33,7 @@ func StartSyncMirrors(queueHandle func(data ...queue.Data) []queue.Data) {
 	if !setting.Mirror.Enabled {
 		return
 	}
-	mirrorQueue = queue.CreateUniqueQueue("mirror", queueHandle, new(SyncRequest))
+	mirrorQueue = queue.CreateUniqueQueue(queue.MirrorQueueName, queueHandle, new(SyncRequest))
 
 	go graceful.GetManager().RunWithShutdownFns(mirrorQueue.Run)
 }

modules/notification/ui/ui.go

@@ -37,7 +37,7 @@ var _ base.Notifier = &notificationService{}
 // NewNotifier create a new notificationService notifier
 func NewNotifier() base.Notifier {
 	ns := &notificationService{}
-	ns.issueQueue = queue.CreateQueue("notification-service", ns.handle, issueNotificationOpts{})
+	ns.issueQueue = queue.CreateQueue(queue.NotificationQueueName, ns.handle, issueNotificationOpts{})
 	return ns
 }
 

modules/queue/setting.go

@@ -12,6 +12,52 @@ import (
 	"code.gitea.io/gitea/modules/setting"
 )
 
+// UniqueQueueName represents an expected name for an UniqueQueue
+type UniqueQueueName string
+
+// list of all expected UniqueQueues
+const (
+	CodeIndexerQueueName     UniqueQueueName = "code_indexer"
+	RepoStatsUpdateQueueName UniqueQueueName = "repo_stats_update"
+	MirrorQueueName          UniqueQueueName = "mirror"
+	PRPatchQueueName         UniqueQueueName = "pr_patch_checker"
+	RepoArchiveQueueName     UniqueQueueName = "repo-archive"
+	PRAutoMergeQueueName     UniqueQueueName = "pr_auto_merge"
+	WebhookDeliveryQueueName UniqueQueueName = "webhook_sender"
+	ActionsJobQueueName      UniqueQueueName = "actions_ready_job"
+)
+
+// KnownUniqueQueueNames represents the list of expected unique queues
+var KnownUniqueQueueNames = []UniqueQueueName{
+	CodeIndexerQueueName,
+	RepoStatsUpdateQueueName,
+	MirrorQueueName,
+	PRPatchQueueName,
+	RepoArchiveQueueName,
+	ActionsJobQueueName,
+}
+
+// QueueName represents an expected name for Queue
+type QueueName string //nolint // allow this to stutter
+
+// list of all expected Queues
+const (
+	IssueIndexerQueueName QueueName = "issue_indexer"
+	NotificationQueueName QueueName = "notification-service"
+	MailerQueueName       QueueName = "mail"
+	PushUpdateQueueName   QueueName = "push_update"
+	TaskQueueName         QueueName = "task"
+)
+
+// KnownQueueNames represents the list of expected queues
+var KnownQueueNames = []QueueName{
+	IssueIndexerQueueName,
+	NotificationQueueName,
+	MailerQueueName,
+	PushUpdateQueueName,
+	TaskQueueName,
+}
+
 func validType(t string) (Type, error) {
 	if len(t) == 0 {
 		return PersistableChannelQueueType, nil
@@ -36,8 +82,19 @@ func getQueueSettings(name string) (setting.QueueSettings, []byte) {
 }
 
 // CreateQueue for name with provided handler and exemplar
-func CreateQueue(name string, handle HandlerFunc, exemplar interface{}) Queue {
-	q, cfg := getQueueSettings(name)
+func CreateQueue(name QueueName, handle HandlerFunc, exemplar interface{}) Queue {
+	found := false
+	for _, expected := range KnownQueueNames {
+		if name == expected {
+			found = true
+			break
+		}
+	}
+	if !found {
+		log.Warn("%s is not an expected name for an Queue", name)
+	}
+
+	q, cfg := getQueueSettings(string(name))
 	if len(cfg) == 0 {
 		return nil
 	}
@@ -57,7 +114,7 @@ func CreateQueue(name string, handle HandlerFunc, exemplar interface{}) Queue {
 			MaxAttempts: q.MaxAttempts,
 			Config:      cfg,
 			QueueLength: q.QueueLength,
-			Name:        name,
+			Name:        string(name),
 		}, exemplar)
 	}
 	if err != nil {
@@ -78,8 +135,19 @@ func CreateQueue(name string, handle HandlerFunc, exemplar interface{}) Queue {
 }
 
 // CreateUniqueQueue for name with provided handler and exemplar
-func CreateUniqueQueue(name string, handle HandlerFunc, exemplar interface{}) UniqueQueue {
-	q, cfg := getQueueSettings(name)
+func CreateUniqueQueue(name UniqueQueueName, handle HandlerFunc, exemplar interface{}) UniqueQueue {
+	found := false
+	for _, expected := range KnownUniqueQueueNames {
+		if name == expected {
+			found = true
+			break
+		}
+	}
+	if !found {
+		log.Warn("%s is not an expected name for an UniqueQueue", name)
+	}
+
+	q, cfg := getQueueSettings(string(name))
 	if len(cfg) == 0 {
 		return nil
 	}
@@ -106,6 +174,7 @@ func CreateUniqueQueue(name string, handle HandlerFunc, exemplar interface{}) Un
 			MaxAttempts: q.MaxAttempts,
 			Config:      cfg,
 			QueueLength: q.QueueLength,
+			Name:        string(name),
 		}, exemplar)
 	}
 	if err != nil {