fix test

Author: wxiaoguang

modules/queue/manager_test.go

@@ -60,7 +60,7 @@ MAX_WORKERS = 2
 
 	q2 := createWorkerPoolQueue("sub", cfgProvider, func(s ...int) (unhandled []int) { return nil }, false)
 	assert.Equal(t, "sub", q2.GetName())
-	assert.Equal(t, "levelqueue", q2.GetType()) // no handler
+	assert.Equal(t, "level", q2.GetType()) // no handler
 	assert.Equal(t, filepath.Join(setting.AppDataPath, "queues-dir2"), q2.baseConfig.DataFullDir)
 	assert.Equal(t, 102, q2.baseConfig.Length)
 	assert.Equal(t, 22, q2.batchLength)

modules/queue/workergroup.go

@@ -6,6 +6,7 @@ package queue
 import (
 	"context"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"code.gitea.io/gitea/modules/log"
@@ -18,9 +19,13 @@ var (
 
 var (
 	workerIdleDuration           = 1 * time.Second
-	unhandledItemRequeueDuration = 1 * time.Second
+	unhandledItemRequeueDuration atomic.Int64 // to avoid data race during test
 )
 
+func init() {
+	unhandledItemRequeueDuration.Store(int64(5 * time.Second))
+}
+
 // workerGroup is a group of workers to work with a WorkerPoolQueue
 type workerGroup[T any] struct {
 	q  *WorkerPoolQueue[T]
@@ -93,9 +98,9 @@ func (q *WorkerPoolQueue[T]) doWorkerHandle(batch []T) {
 	unhandled := q.safeHandler(batch...)
 	// if none of the items were handled, it should back-off for a few seconds
 	// in this case the handler (eg: document indexer) may have encountered some errors/failures
-	if len(unhandled) == len(batch) && unhandledItemRequeueDuration != 0 {
+	if len(unhandled) == len(batch) && unhandledItemRequeueDuration.Load() != 0 {
 		log.Error("Queue %q failed to handle batch of %d items, backoff for a few seconds", q.GetName(), len(batch))
-		time.Sleep(unhandledItemRequeueDuration)
+		time.Sleep(time.Duration(unhandledItemRequeueDuration.Load()))
 	}
 	for _, item := range unhandled {
 		if err := q.Push(item); err != nil {

modules/queue/workerqueue_test.go

@@ -31,11 +31,9 @@ func runWorkerPoolQueue[T any](q *WorkerPoolQueue[T]) func() {
 }
 
 func TestWorkerPoolQueueUnhandled(t *testing.T) {
-	oldUnhandledItemRequeueDuration := unhandledItemRequeueDuration
-	unhandledItemRequeueDuration = 0
-	defer func() {
-		unhandledItemRequeueDuration = oldUnhandledItemRequeueDuration
-	}()
+	oldUnhandledItemRequeueDuration := unhandledItemRequeueDuration.Load()
+	unhandledItemRequeueDuration.Store(0)
+	defer unhandledItemRequeueDuration.Store(oldUnhandledItemRequeueDuration)
 
 	mu := sync.Mutex{}