Merge remote-tracking branch 'origin/main' into zzc/dev/act_runner_exec

modules/queue/backoff.go

@@ -8,7 +8,7 @@ import (
 	"time"
 )
 
-const (
+var (
 	backoffBegin = 50 * time.Millisecond
 	backoffUpper = 2 * time.Second
 )
@@ -18,6 +18,14 @@ type (
 	backoffFuncErr           func() (retry bool, err error)
 )
 
+func mockBackoffDuration(d time.Duration) func() {
+	oldBegin, oldUpper := backoffBegin, backoffUpper
+	backoffBegin, backoffUpper = d, d
+	return func() {
+		backoffBegin, backoffUpper = oldBegin, oldUpper
+	}
+}
+
 func backoffRetErr[T any](ctx context.Context, begin, upper time.Duration, end <-chan time.Time, fn backoffFuncRetErr[T]) (ret T, err error) {
 	d := begin
 	for {

modules/queue/base_channel.go

@@ -120,7 +120,7 @@ func (q *baseChannel) RemoveAll(ctx context.Context) error {
 	q.mu.Lock()
 	defer q.mu.Unlock()
 
-	for q.c != nil && len(q.c) > 0 {
+	for len(q.c) > 0 {
 		<-q.c
 	}
 

modules/queue/base_levelqueue_common.go

@@ -83,7 +83,7 @@ func prepareLevelDB(cfg *BaseConfig) (conn string, db *leveldb.DB, err error) {
 		}
 		conn = cfg.ConnStr
 	}
-	for i := 0; i < 10; i++ {
+	for range 10 {
 		if db, err = nosql.GetManager().GetLevelDB(conn); err == nil {
 			break
 		}

modules/queue/base_levelqueue_test.go

@@ -11,6 +11,7 @@ import (
 
 	"gitea.com/lunny/levelqueue"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"github.com/syndtr/goleveldb/leveldb"
 )
 
@@ -29,9 +30,7 @@ func TestCorruptedLevelQueue(t *testing.T) {
 	// sometimes the levelqueue could be in a corrupted state, this test is to make sure it can recover from it
 	dbDir := t.TempDir() + "/levelqueue-test"
 	db, err := leveldb.OpenFile(dbDir, nil)
-	if !assert.NoError(t, err) {
-		return
-	}
+	require.NoError(t, err)
 	defer db.Close()
 
 	assert.NoError(t, db.Put([]byte("other-key"), []byte("other-value"), nil))

modules/queue/base_redis.go

@@ -29,7 +29,7 @@ func newBaseRedisGeneric(cfg *BaseConfig, unique bool) (baseQueue, error) {
 	client := nosql.GetManager().GetRedisClient(cfg.ConnStr)
 
 	var err error
-	for i := 0; i < 10; i++ {
+	for range 10 {
 		err = client.Ping(graceful.GetManager().ShutdownContext()).Err()
 		if err == nil {
 			break

modules/queue/base_redis_test.go

@@ -14,6 +14,7 @@ import (
 	"code.gitea.io/gitea/modules/setting"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func waitRedisReady(conn string, dur time.Duration) (ready bool) {
@@ -61,9 +62,7 @@ func TestBaseRedis(t *testing.T) {
 			return
 		}
 		assert.NoError(t, redisServer.Start())
-		if !assert.True(t, waitRedisReady("redis://redis:6379/0", 5*time.Second), "start redis-server") {
-			return
-		}
+		require.True(t, waitRedisReady("redis://redis:6379/0", 5*time.Second), "start redis-server")
 	}
 
 	testQueueBasic(t, newBaseRedisSimple, toBaseConfig("baseRedis", setting.QueueSettings{Length: 10, ConnStr: "redis://redis:6379/0"}), false)

modules/queue/base_test.go

@@ -17,11 +17,11 @@ func testQueueBasic(t *testing.T, newFn func(cfg *BaseConfig) (baseQueue, error)
 		q, err := newFn(cfg)
 		assert.NoError(t, err)
 
-		ctx := context.Background()
+		ctx := t.Context()
 		_ = q.RemoveAll(ctx)
 		cnt, err := q.Len(ctx)
 		assert.NoError(t, err)
-		assert.EqualValues(t, 0, cnt)
+		assert.Equal(t, 0, cnt)
 
 		// push the first item
 		err = q.PushItem(ctx, []byte("foo"))
@@ -29,7 +29,7 @@ func testQueueBasic(t *testing.T, newFn func(cfg *BaseConfig) (baseQueue, error)
 
 		cnt, err = q.Len(ctx)
 		assert.NoError(t, err)
-		assert.EqualValues(t, 1, cnt)
+		assert.Equal(t, 1, cnt)
 
 		// push a duplicate item
 		err = q.PushItem(ctx, []byte("foo"))
@@ -45,11 +45,11 @@ func testQueueBasic(t *testing.T, newFn func(cfg *BaseConfig) (baseQueue, error)
 		has, err := q.HasItem(ctx, []byte("foo"))
 		assert.NoError(t, err)
 		if !isUnique {
-			assert.EqualValues(t, 2, cnt)
-			assert.EqualValues(t, false, has) // non-unique queues don't check for duplicates
+			assert.Equal(t, 2, cnt)
+			assert.False(t, has) // non-unique queues don't check for duplicates
 		} else {
-			assert.EqualValues(t, 1, cnt)
-			assert.EqualValues(t, true, has)
+			assert.Equal(t, 1, cnt)
+			assert.True(t, has)
 		}
 
 		// push another item
@@ -59,18 +59,18 @@ func testQueueBasic(t *testing.T, newFn func(cfg *BaseConfig) (baseQueue, error)
 		// pop the first item (and the duplicate if non-unique)
 		it, err := q.PopItem(ctx)
 		assert.NoError(t, err)
-		assert.EqualValues(t, "foo", string(it))
+		assert.Equal(t, "foo", string(it))
 
 		if !isUnique {
 			it, err = q.PopItem(ctx)
 			assert.NoError(t, err)
-			assert.EqualValues(t, "foo", string(it))
+			assert.Equal(t, "foo", string(it))
 		}
 
 		// pop another item
 		it, err = q.PopItem(ctx)
 		assert.NoError(t, err)
-		assert.EqualValues(t, "bar", string(it))
+		assert.Equal(t, "bar", string(it))
 
 		// pop an empty queue (timeout, cancel)
 		ctxTimed, cancel := context.WithTimeout(ctx, 10*time.Millisecond)
@@ -87,7 +87,7 @@ func testQueueBasic(t *testing.T, newFn func(cfg *BaseConfig) (baseQueue, error)
 
 		// test blocking push if queue is full
 		for i := 0; i < cfg.Length; i++ {
-			err = q.PushItem(ctx, []byte(fmt.Sprintf("item-%d", i)))
+			err = q.PushItem(ctx, fmt.Appendf(nil, "item-%d", i))
 			assert.NoError(t, err)
 		}
 		ctxTimed, cancel = context.WithTimeout(ctx, 10*time.Millisecond)
@@ -101,19 +101,19 @@ func testQueueBasic(t *testing.T, newFn func(cfg *BaseConfig) (baseQueue, error)
 		pushBlockTime = 30 * time.Millisecond
 		err = q.PushItem(ctx, []byte("item-full"))
 		assert.ErrorIs(t, err, context.DeadlineExceeded)
-		assert.True(t, time.Since(timeStart) >= pushBlockTime*2/3)
+		assert.GreaterOrEqual(t, time.Since(timeStart), pushBlockTime*2/3)
 		pushBlockTime = oldPushBlockTime
 
 		// remove all
 		cnt, err = q.Len(ctx)
 		assert.NoError(t, err)
-		assert.EqualValues(t, cfg.Length, cnt)
+		assert.Equal(t, cfg.Length, cnt)
 
 		_ = q.RemoveAll(ctx)
 
 		cnt, err = q.Len(ctx)
 		assert.NoError(t, err)
-		assert.EqualValues(t, 0, cnt)
+		assert.Equal(t, 0, cnt)
 	})
 }
 
@@ -121,12 +121,12 @@ func TestBaseDummy(t *testing.T) {
 	q, err := newBaseDummy(&BaseConfig{}, true)
 	assert.NoError(t, err)
 
-	ctx := context.Background()
+	ctx := t.Context()
 	assert.NoError(t, q.PushItem(ctx, []byte("foo")))
 
 	cnt, err := q.Len(ctx)
 	assert.NoError(t, err)
-	assert.EqualValues(t, 0, cnt)
+	assert.Equal(t, 0, cnt)
 
 	has, err := q.HasItem(ctx, []byte("foo"))
 	assert.NoError(t, err)

modules/queue/manager.go

@@ -5,6 +5,8 @@ package queue
 
 import (
 	"context"
+	"errors"
+	"maps"
 	"sync"
 	"time"
 
@@ -32,6 +34,7 @@ type ManagedWorkerPoolQueue interface {
 
 	// FlushWithContext tries to make the handler process all items in the queue synchronously.
 	// It is for testing purpose only. It's not designed to be used in a cluster.
+	// Negative timeout means discarding all items in the queue.
 	FlushWithContext(ctx context.Context, timeout time.Duration) error
 
 	// RemoveAllItems removes all items in the base queue (on-the-fly items are not affected)
@@ -68,23 +71,22 @@ func (m *Manager) ManagedQueues() map[int64]ManagedWorkerPoolQueue {
 	defer m.mu.Unlock()
 
 	queues := make(map[int64]ManagedWorkerPoolQueue, len(m.Queues))
-	for k, v := range m.Queues {
-		queues[k] = v
-	}
+	maps.Copy(queues, m.Queues)
 	return queues
 }
 
 // FlushAll tries to make all managed queues process all items synchronously, until timeout or the queue is empty.
 // It is for testing purpose only. It's not designed to be used in a cluster.
+// Negative timeout means discarding all items in the queue.
 func (m *Manager) FlushAll(ctx context.Context, timeout time.Duration) error {
-	var finalErr error
+	var finalErrors []error
 	qs := m.ManagedQueues()
 	for _, q := range qs {
 		if err := q.FlushWithContext(ctx, timeout); err != nil {
-			finalErr = err // TODO: in Go 1.20: errors.Join
+			finalErrors = append(finalErrors, err)
 		}
 	}
-	return finalErr
+	return errors.Join(finalErrors...)
 }
 
 // CreateSimpleQueue creates a simple queue from global setting config provider by name

modules/queue/manager_test.go

@@ -4,7 +4,6 @@
 package queue
 
 import (
-	"context"
 	"path/filepath"
 	"testing"
 
@@ -46,9 +45,9 @@ CONN_STR = redis://
 	assert.Equal(t, "default", q.GetName())
 	assert.Equal(t, "level", q.GetType())
 	assert.Equal(t, filepath.Join(setting.AppDataPath, "queues/common"), q.baseConfig.DataFullDir)
-	assert.Equal(t, 100, q.baseConfig.Length)
+	assert.Equal(t, 100000, q.baseConfig.Length)
 	assert.Equal(t, 20, q.batchLength)
-	assert.Equal(t, "", q.baseConfig.ConnStr)
+	assert.Empty(t, q.baseConfig.ConnStr)
 	assert.Equal(t, "default_queue", q.baseConfig.QueueFullName)
 	assert.Equal(t, "default_queue_unique", q.baseConfig.SetFullName)
 	assert.NotZero(t, q.GetWorkerMaxNumber())
@@ -80,7 +79,7 @@ MAX_WORKERS = 123
 
 	assert.NoError(t, err)
 
-	q1 := createWorkerPoolQueue[string](context.Background(), "no-such", cfgProvider, nil, false)
+	q1 := createWorkerPoolQueue[string](t.Context(), "no-such", cfgProvider, nil, false)
 	assert.Equal(t, "no-such", q1.GetName())
 	assert.Equal(t, "dummy", q1.GetType()) // no handler, so it becomes dummy
 	assert.Equal(t, filepath.Join(setting.AppDataPath, "queues/dir1"), q1.baseConfig.DataFullDir)
@@ -96,13 +95,13 @@ MAX_WORKERS = 123
 	assert.Equal(t, "string", q1.GetItemTypeName())
 	qid1 := GetManager().qidCounter
 
-	q2 := createWorkerPoolQueue(context.Background(), "sub", cfgProvider, func(s ...int) (unhandled []int) { return nil }, false)
+	q2 := createWorkerPoolQueue(t.Context(), "sub", cfgProvider, func(s ...int) (unhandled []int) { return nil }, false)
 	assert.Equal(t, "sub", q2.GetName())
 	assert.Equal(t, "level", q2.GetType())
 	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)
-	assert.Equal(t, "", q2.baseConfig.ConnStr)
+	assert.Empty(t, q2.baseConfig.ConnStr)
 	assert.Equal(t, "sub_q2", q2.baseConfig.QueueFullName)
 	assert.Equal(t, "sub_q2_u2", q2.baseConfig.SetFullName)
 	assert.Equal(t, 123, q2.GetWorkerMaxNumber())
@@ -118,7 +117,7 @@ MAX_WORKERS = 123
 	assert.Equal(t, 120, q1.workerMaxNum)
 
 	stop := runWorkerPoolQueue(q2)
-	assert.NoError(t, GetManager().GetManagedQueue(qid2).FlushWithContext(context.Background(), 0))
-	assert.NoError(t, GetManager().FlushAll(context.Background(), 0))
+	assert.NoError(t, GetManager().GetManagedQueue(qid2).FlushWithContext(t.Context(), 0))
+	assert.NoError(t, GetManager().FlushAll(t.Context(), 0))
 	stop()
 }

modules/queue/queue.go

@@ -1,27 +1,64 @@
 // Copyright 2023 The Gitea Authors. All rights reserved.
 // SPDX-License-Identifier: MIT
 
-// Package queue implements a specialized queue system for Gitea.
+// Package queue implements a specialized concurrent queue system for Gitea.
 //
-// There are two major kinds of concepts:
+// Terminology:
 //
-// * The "base queue": channel, level, redis:
-//   - They have the same abstraction, the same interface, and they are tested by the same testing code.
-//   - The dummy(immediate) queue is special, it's not a real queue, it's only used as a no-op queue or a testing queue.
+//  1. Item:
+//     - An item can be a simple value, such as an integer, or a more complex structure that has multiple fields.
+//     Usually a item serves as a task or a message. Sets of items will be sent to a queue handler to be processed.
+//     - It's represented as a JSON-marshaled binary slice in the queue
+//     - Since the item is marshaled by JSON, and JSON doesn't have stable key-order/type support,
+//     so the decoded handler item may not be the same as the original "pushed" one if you use map/any types,
 //
-// * The WorkerPoolQueue: it uses the "base queue" to provide "worker pool" function.
-//   - It calls the "handler" to process the data in the base queue.
-//   - Its "Push" function doesn't block forever,
-//     it will return an error if the queue is full after the timeout.
+//  2. Batch:
+//     - A collection of items that are grouped together for processing. Each worker receives a batch of items.
+//
+//  3. Worker:
+//     - Individual unit of execution designed to process items from the queue. It's a goroutine that calls the Handler.
+//     - Workers will get new items through a channel (WorkerPoolQueue is responsible for the distribution).
+//     - Workers operate in parallel. The default value of max workers is determined by the setting system.
+//
+//  4. Handler (represented by HandlerFuncT type):
+//     - It's the function responsible for processing items. Each active worker will call it.
+//     - If an item or some items are not psuccessfully rocessed, the handler could return them as "unhandled items".
+//     In such scenarios, the queue system ensures these unhandled items are returned to the base queue after a brief delay.
+//     This mechanism is particularly beneficial in cases where the processing entity (like a document indexer) is
+//     temporarily unavailable. It ensures that no item is skipped or lost due to transient failures in the processing
+//     mechanism.
+//
+//  5. Base queue:
+//     - Represents the underlying storage mechanism for the queue. There are several implementations:
+//     - Channel: Uses Go's native channel constructs to manage the queue, suitable for in-memory queuing.
+//     - LevelDB: Especially useful in persistent queues for single instances.
+//     - Redis: Suitable for clusters, where we may have multiple nodes.
+//     - Dummy: This is special, it's not a real queue, it's a immediate no-op queue, which is useful for tests.
+//     - They all have the same abstraction, the same interface, and they are tested by the same testing code.
+//
+// 6. WorkerPoolQueue:
+//   - It's responsible to glue all together, using the "base queue" to provide "worker pool" functionality. It creates
+//     new workers if needed and can flush the queue, running all the items synchronously till it finishes.
+//   - Its "Push" function doesn't block forever, it will return an error if the queue is full after the timeout.
+//
+// 7. Manager:
+//   - The purpose of it is to serve as a centralized manager for multiple WorkerPoolQueue instances. Whenever we want
+//     to create a new queue, flush, or get a specific queue, we could use it.
 //
 // A queue can be "simple" or "unique". A unique queue will try to avoid duplicate items.
 // Unique queue's "Has" function can be used to check whether an item is already in the queue,
-// although it's not 100% reliable due to there is no proper transaction support.
+// although it's not 100% reliable due to the lack of proper transaction support.
 // Simple queue's "Has" function always returns "has=false".
 //
-// The HandlerFuncT function is called by the WorkerPoolQueue to process the data in the base queue.
-// If the handler returns "unhandled" items, they will be re-queued to the base queue after a slight delay,
-// in case the item processor (eg: document indexer) is not available.
+// A WorkerPoolQueue is a generic struct; this means it will work with any type but just for that type.
+// If you want another kind of items to run, you would have to call the manager to create a new WorkerPoolQueue for you
+// with a different handler that works with this new type of item. As an example of this:
+//
+//	 func Init() error {
+//		 itemQueue = queue.CreateSimpleQueue(graceful.GetManager().ShutdownContext(), "queue-name", handler)
+//		 ...
+//	 }
+//	 func handler(items ...*mypkg.QueueItem) []*mypkg.QueueItem { ... }
 package queue
 
 import "code.gitea.io/gitea/modules/util"

modules/queue/workergroup.go

@@ -23,7 +23,7 @@ var (
 )
 
 func init() {
-	unhandledItemRequeueDuration.Store(int64(5 * time.Second))
+	unhandledItemRequeueDuration.Store(int64(time.Second))
 }
 
 // workerGroup is a group of workers to work with a WorkerPoolQueue
@@ -60,6 +60,11 @@ func (q *WorkerPoolQueue[T]) doDispatchBatchToWorker(wg *workerGroup[T], flushCh
 		full = true
 	}
 
+	// TODO: the logic could be improved in the future, to avoid a data-race between "doStartNewWorker" and "workerNum"
+	// The root problem is that if we skip "doStartNewWorker" here, the "workerNum" might be decreased by other workers later
+	// So ideally, it should check whether there are enough workers by some approaches, and start new workers if necessary.
+	// This data-race is not serious, as long as a new worker will be started soon to make sure there are enough workers,
+	// so no need to hugely refactor at the moment.
 	q.workerNumMu.Lock()
 	noWorker := q.workerNum == 0
 	if full || noWorker {
@@ -99,7 +104,12 @@ func (q *WorkerPoolQueue[T]) doWorkerHandle(batch []T) {
 	// 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.Load() != 0 {
+		if q.isFlushing.Load() {
+			return // do not requeue items when flushing, since all items failed, requeue them will continue failing.
+		}
 		log.Error("Queue %q failed to handle batch of %d items, backoff for a few seconds", q.GetName(), len(batch))
+		// TODO: ideally it shouldn't "sleep" here (blocks the worker, then blocks flush).
+		// It could debounce the requeue operation, and try to requeue the items in the future.
 		select {
 		case <-q.ctxRun.Done():
 		case <-time.After(time.Duration(unhandledItemRequeueDuration.Load())):
@@ -133,6 +143,14 @@ func (q *WorkerPoolQueue[T]) basePushForShutdown(items ...T) bool {
 	return true
 }
 
+func resetIdleTicker(t *time.Ticker, dur time.Duration) {
+	t.Reset(dur)
+	select {
+	case <-t.C:
+	default:
+	}
+}
+
 // doStartNewWorker starts a new worker for the queue, the worker reads from worker's channel and handles the items.
 func (q *WorkerPoolQueue[T]) doStartNewWorker(wp *workerGroup[T]) {
 	wp.wg.Add(1)
@@ -144,6 +162,8 @@ func (q *WorkerPoolQueue[T]) doStartNewWorker(wp *workerGroup[T]) {
 		defer log.Debug("Queue %q stops idle worker", q.GetName())
 
 		t := time.NewTicker(workerIdleDuration)
+		defer t.Stop()
+
 		keepWorking := true
 		stopWorking := func() {
 			q.workerNumMu.Lock()
@@ -158,13 +178,14 @@ func (q *WorkerPoolQueue[T]) doStartNewWorker(wp *workerGroup[T]) {
 			case batch, ok := <-q.batchChan:
 				if !ok {
 					stopWorking()
-				} else {
-					q.doWorkerHandle(batch)
-					t.Reset(workerIdleDuration)
+					continue
 				}
+				q.doWorkerHandle(batch)
+				// reset the idle ticker, and drain the tick after reset in case a tick is already triggered
+				resetIdleTicker(t, workerIdleDuration) // key code for TestWorkerPoolQueueWorkerIdleReset
 			case <-t.C:
 				q.workerNumMu.Lock()
-				keepWorking = q.workerNum <= 1
+				keepWorking = q.workerNum <= 1 // keep the last worker running
 				if !keepWorking {
 					q.workerNum--
 				}
@@ -177,19 +198,37 @@ func (q *WorkerPoolQueue[T]) doStartNewWorker(wp *workerGroup[T]) {
 // doFlush flushes the queue: it tries to read all items from the queue and handles them.
 // It is for testing purpose only. It's not designed to work for a cluster.
 func (q *WorkerPoolQueue[T]) doFlush(wg *workerGroup[T], flush flushType) {
+	q.isFlushing.Store(true)
+	defer q.isFlushing.Store(false)
+
 	log.Debug("Queue %q starts flushing", q.GetName())
 	defer log.Debug("Queue %q finishes flushing", q.GetName())
 
 	// stop all workers, and prepare a new worker context to start new workers
-
 	wg.ctxWorkerCancel()
 	wg.wg.Wait()
 
 	defer func() {
-		close(flush)
+		close(flush.c)
 		wg.doPrepareWorkerContext()
 	}()
 
+	if flush.timeout < 0 {
+		// discard everything
+		wg.batchBuffer = nil
+		for {
+			select {
+			case <-wg.popItemChan:
+			case <-wg.popItemErr:
+			case <-q.batchChan:
+			case <-q.ctxRun.Done():
+				return
+			default:
+				return
+			}
+		}
+	}
+
 	// drain the batch channel first
 loop:
 	for {
@@ -205,6 +244,9 @@ loop:
 	emptyCounter := 0
 	for {
 		select {
+		case <-q.ctxRun.Done():
+			log.Debug("Queue %q is shutting down", q.GetName())
+			return
 		case data, dataOk := <-wg.popItemChan:
 			if !dataOk {
 				return
@@ -220,9 +262,6 @@ loop:
 				log.Error("Failed to pop item from queue %q (doFlush): %v", q.GetName(), err)
 			}
 			return
-		case <-q.ctxRun.Done():
-			log.Debug("Queue %q is shutting down", q.GetName())
-			return
 		case <-time.After(20 * time.Millisecond):
 			// There is no reliable way to make sure all queue items are consumed by the Flush, there always might be some items stored in some buffers/temp variables.
 			// If we run Gitea in a cluster, we can even not guarantee all items are consumed in a deterministic instance.
@@ -300,6 +339,15 @@ func (q *WorkerPoolQueue[T]) doRun() {
 	var batchDispatchC <-chan time.Time = infiniteTimerC
 	for {
 		select {
+		case flush := <-q.flushChan:
+			// before flushing, it needs to try to dispatch the batch to worker first, in case there is no worker running
+			// after the flushing, there is at least one worker running, so "doFlush" could wait for workers to finish
+			// since we are already in a "flush" operation, so the dispatching function shouldn't read the flush chan.
+			q.doDispatchBatchToWorker(wg, skipFlushChan)
+			q.doFlush(wg, flush)
+		case <-q.ctxRun.Done():
+			log.Debug("Queue %q is shutting down", q.GetName())
+			return
 		case data, dataOk := <-wg.popItemChan:
 			if !dataOk {
 				return
@@ -318,20 +366,11 @@ func (q *WorkerPoolQueue[T]) doRun() {
 		case <-batchDispatchC:
 			batchDispatchC = infiniteTimerC
 			q.doDispatchBatchToWorker(wg, q.flushChan)
-		case flush := <-q.flushChan:
-			// before flushing, it needs to try to dispatch the batch to worker first, in case there is no worker running
-			// after the flushing, there is at least one worker running, so "doFlush" could wait for workers to finish
-			// since we are already in a "flush" operation, so the dispatching function shouldn't read the flush chan.
-			q.doDispatchBatchToWorker(wg, skipFlushChan)
-			q.doFlush(wg, flush)
 		case err := <-wg.popItemErr:
 			if !q.isCtxRunCanceled() {
 				log.Error("Failed to pop item from queue %q (doRun): %v", q.GetName(), err)
 			}
 			return
-		case <-q.ctxRun.Done():
-			log.Debug("Queue %q is shutting down", q.GetName())
-			return
 		}
 	}
 }

modules/queue/workerqueue.go

@@ -6,6 +6,7 @@ package queue
 import (
 	"context"
 	"fmt"
+	"runtime/pprof"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -32,8 +33,9 @@ type WorkerPoolQueue[T any] struct {
 	baseConfig    *BaseConfig
 	baseQueue     baseQueue
 
-	batchChan chan []T
-	flushChan chan flushType
+	batchChan  chan []T
+	flushChan  chan flushType
+	isFlushing atomic.Bool
 
 	batchLength     int
 	workerNum       int
@@ -42,7 +44,10 @@ type WorkerPoolQueue[T any] struct {
 	workerNumMu     sync.Mutex
 }
 
-type flushType chan struct{}
+type flushType struct {
+	timeout time.Duration
+	c       chan struct{}
+}
 
 var _ ManagedWorkerPoolQueue = (*WorkerPoolQueue[any])(nil)
 
@@ -104,12 +109,12 @@ func (q *WorkerPoolQueue[T]) FlushWithContext(ctx context.Context, timeout time.
 	if timeout > 0 {
 		after = time.After(timeout)
 	}
-	c := make(flushType)
+	flush := flushType{timeout: timeout, c: make(chan struct{})}
 
 	// send flush request
 	// if it blocks, it means that there is a flush in progress or the queue hasn't been started yet
 	select {
-	case q.flushChan <- c:
+	case q.flushChan <- flush:
 	case <-ctx.Done():
 		return ctx.Err()
 	case <-q.ctxRun.Done():
@@ -120,7 +125,7 @@ func (q *WorkerPoolQueue[T]) FlushWithContext(ctx context.Context, timeout time.
 
 	// wait for flush to finish
 	select {
-	case <-c:
+	case <-flush.c:
 		return nil
 	case <-ctx.Done():
 		return ctx.Err()
@@ -237,6 +242,9 @@ func NewWorkerPoolQueueWithContext[T any](ctx context.Context, name string, queu
 	w.origHandler = handler
 	w.safeHandler = func(t ...T) (unhandled []T) {
 		defer func() {
+			// FIXME: there is no ctx support in the handler, so process manager is unable to restore the labels
+			// so here we explicitly set the "queue ctx" labels again after the handler is done
+			pprof.SetGoroutineLabels(w.ctxRun)
 			err := recover()
 			if err != nil {
 				log.Error("Recovered from panic in queue %q handler: %v\n%s", name, err, log.Stack(2))

modules/queue/workerqueue_test.go

@@ -4,13 +4,15 @@
 package queue
 
 import (
-	"context"
+	"slices"
 	"strconv"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"code.gitea.io/gitea/modules/setting"
+	"code.gitea.io/gitea/modules/test"
 
 	"github.com/stretchr/testify/assert"
 )
@@ -55,15 +57,15 @@ func TestWorkerPoolQueueUnhandled(t *testing.T) {
 			testRecorder.Record("push:%v", i)
 			assert.NoError(t, q.Push(i))
 		}
-		assert.NoError(t, q.FlushWithContext(context.Background(), 0))
+		assert.NoError(t, q.FlushWithContext(t.Context(), 0))
 		stop()
 
 		ok := true
 		for i := 0; i < queueSetting.Length; i++ {
 			if i%2 == 0 {
-				ok = ok && assert.EqualValues(t, 2, m[i], "test %s: item %d", t.Name(), i)
+				ok = ok && assert.Equal(t, 2, m[i], "test %s: item %d", t.Name(), i)
 			} else {
-				ok = ok && assert.EqualValues(t, 1, m[i], "test %s: item %d", t.Name(), i)
+				ok = ok && assert.Equal(t, 1, m[i], "test %s: item %d", t.Name(), i)
 			}
 		}
 		if !ok {
@@ -75,17 +77,17 @@ func TestWorkerPoolQueueUnhandled(t *testing.T) {
 
 	runCount := 2 // we can run these tests even hundreds times to see its stability
 	t.Run("1/1", func(t *testing.T) {
-		for i := 0; i < runCount; i++ {
+		for range runCount {
 			test(t, setting.QueueSettings{BatchLength: 1, MaxWorkers: 1})
 		}
 	})
 	t.Run("3/1", func(t *testing.T) {
-		for i := 0; i < runCount; i++ {
+		for range runCount {
 			test(t, setting.QueueSettings{BatchLength: 3, MaxWorkers: 1})
 		}
 	})
 	t.Run("4/5", func(t *testing.T) {
-		for i := 0; i < runCount; i++ {
+		for range runCount {
 			test(t, setting.QueueSettings{BatchLength: 4, MaxWorkers: 5})
 		}
 	})
@@ -94,17 +96,17 @@ func TestWorkerPoolQueueUnhandled(t *testing.T) {
 func TestWorkerPoolQueuePersistence(t *testing.T) {
 	runCount := 2 // we can run these tests even hundreds times to see its stability
 	t.Run("1/1", func(t *testing.T) {
-		for i := 0; i < runCount; i++ {
+		for range runCount {
 			testWorkerPoolQueuePersistence(t, setting.QueueSettings{BatchLength: 1, MaxWorkers: 1, Length: 100})
 		}
 	})
 	t.Run("3/1", func(t *testing.T) {
-		for i := 0; i < runCount; i++ {
+		for range runCount {
 			testWorkerPoolQueuePersistence(t, setting.QueueSettings{BatchLength: 3, MaxWorkers: 1, Length: 100})
 		}
 	})
 	t.Run("4/5", func(t *testing.T) {
-		for i := 0; i < runCount; i++ {
+		for range runCount {
 			testWorkerPoolQueuePersistence(t, setting.QueueSettings{BatchLength: 4, MaxWorkers: 5, Length: 100})
 		}
 	})
@@ -139,7 +141,7 @@ func testWorkerPoolQueuePersistence(t *testing.T, queueSetting setting.QueueSett
 
 		q, _ := newWorkerPoolQueueForTest("pr_patch_checker_test", queueSetting, testHandler, true)
 		stop := runWorkerPoolQueue(q)
-		for i := 0; i < testCount; i++ {
+		for i := range testCount {
 			_ = q.Push("task-" + strconv.Itoa(i))
 		}
 		close(startWhenAllReady)
@@ -163,23 +165,19 @@ func testWorkerPoolQueuePersistence(t *testing.T, queueSetting setting.QueueSett
 
 		q, _ := newWorkerPoolQueueForTest("pr_patch_checker_test", queueSetting, testHandler, true)
 		stop := runWorkerPoolQueue(q)
-		assert.NoError(t, q.FlushWithContext(context.Background(), 0))
+		assert.NoError(t, q.FlushWithContext(t.Context(), 0))
 		stop()
 	}
 
 	q2() // restart the queue to continue to execute the tasks in it
 
-	assert.NotZero(t, len(tasksQ1))
-	assert.NotZero(t, len(tasksQ2))
-	assert.EqualValues(t, testCount, len(tasksQ1)+len(tasksQ2))
+	assert.NotEmpty(t, tasksQ1)
+	assert.NotEmpty(t, tasksQ2)
+	assert.Equal(t, testCount, len(tasksQ1)+len(tasksQ2))
 }
 
 func TestWorkerPoolQueueActiveWorkers(t *testing.T) {
-	oldWorkerIdleDuration := workerIdleDuration
-	workerIdleDuration = 300 * time.Millisecond
-	defer func() {
-		workerIdleDuration = oldWorkerIdleDuration
-	}()
+	defer test.MockVariableValue(&workerIdleDuration, 300*time.Millisecond)()
 
 	handler := func(items ...int) (unhandled []int) {
 		time.Sleep(100 * time.Millisecond)
@@ -188,34 +186,34 @@ func TestWorkerPoolQueueActiveWorkers(t *testing.T) {
 
 	q, _ := newWorkerPoolQueueForTest("test-workpoolqueue", setting.QueueSettings{Type: "channel", BatchLength: 1, MaxWorkers: 1, Length: 100}, handler, false)
 	stop := runWorkerPoolQueue(q)
-	for i := 0; i < 5; i++ {
+	for i := range 5 {
 		assert.NoError(t, q.Push(i))
 	}
 
 	time.Sleep(50 * time.Millisecond)
-	assert.EqualValues(t, 1, q.GetWorkerNumber())
-	assert.EqualValues(t, 1, q.GetWorkerActiveNumber())
+	assert.Equal(t, 1, q.GetWorkerNumber())
+	assert.Equal(t, 1, q.GetWorkerActiveNumber())
 	time.Sleep(500 * time.Millisecond)
-	assert.EqualValues(t, 1, q.GetWorkerNumber())
-	assert.EqualValues(t, 0, q.GetWorkerActiveNumber())
+	assert.Equal(t, 1, q.GetWorkerNumber())
+	assert.Equal(t, 0, q.GetWorkerActiveNumber())
 	time.Sleep(workerIdleDuration)
-	assert.EqualValues(t, 1, q.GetWorkerNumber()) // there is at least one worker after the queue begins working
+	assert.Equal(t, 1, q.GetWorkerNumber()) // there is at least one worker after the queue begins working
 	stop()
 
 	q, _ = newWorkerPoolQueueForTest("test-workpoolqueue", setting.QueueSettings{Type: "channel", BatchLength: 1, MaxWorkers: 3, Length: 100}, handler, false)
 	stop = runWorkerPoolQueue(q)
-	for i := 0; i < 15; i++ {
+	for i := range 15 {
 		assert.NoError(t, q.Push(i))
 	}
 
 	time.Sleep(50 * time.Millisecond)
-	assert.EqualValues(t, 3, q.GetWorkerNumber())
-	assert.EqualValues(t, 3, q.GetWorkerActiveNumber())
+	assert.Equal(t, 3, q.GetWorkerNumber())
+	assert.Equal(t, 3, q.GetWorkerActiveNumber())
 	time.Sleep(500 * time.Millisecond)
-	assert.EqualValues(t, 3, q.GetWorkerNumber())
-	assert.EqualValues(t, 0, q.GetWorkerActiveNumber())
+	assert.Equal(t, 3, q.GetWorkerNumber())
+	assert.Equal(t, 0, q.GetWorkerActiveNumber())
 	time.Sleep(workerIdleDuration)
-	assert.EqualValues(t, 1, q.GetWorkerNumber()) // there is at least one worker after the queue begins working
+	assert.Equal(t, 1, q.GetWorkerNumber()) // there is at least one worker after the queue begins working
 	stop()
 }
 
@@ -242,11 +240,45 @@ func TestWorkerPoolQueueShutdown(t *testing.T) {
 	}
 	<-handlerCalled
 	time.Sleep(200 * time.Millisecond) // wait for a while to make sure all workers are active
-	assert.EqualValues(t, 4, q.GetWorkerActiveNumber())
+	assert.Equal(t, 4, q.GetWorkerActiveNumber())
 	stop() // stop triggers shutdown
-	assert.EqualValues(t, 0, q.GetWorkerActiveNumber())
+	assert.Equal(t, 0, q.GetWorkerActiveNumber())
 
 	// no item was ever handled, so we still get all of them again
 	q, _ = newWorkerPoolQueueForTest("test-workpoolqueue", qs, handler, false)
-	assert.EqualValues(t, 20, q.GetQueueItemNumber())
+	assert.Equal(t, 20, q.GetQueueItemNumber())
+}
+
+func TestWorkerPoolQueueWorkerIdleReset(t *testing.T) {
+	defer test.MockVariableValue(&workerIdleDuration, 10*time.Millisecond)()
+	defer mockBackoffDuration(5 * time.Millisecond)()
+
+	var q *WorkerPoolQueue[int]
+	var handledCount atomic.Int32
+	var hasOnlyOneWorkerRunning atomic.Bool
+	handler := func(items ...int) (unhandled []int) {
+		handledCount.Add(int32(len(items)))
+		// make each work have different duration, and check the active worker number periodically
+		var activeNums []int
+		for i := 0; i < 5-items[0]%2; i++ {
+			time.Sleep(workerIdleDuration * 2)
+			activeNums = append(activeNums, q.GetWorkerActiveNumber())
+		}
+		// When the queue never becomes empty, the existing workers should keep working
+		// It is not 100% true at the moment because the data-race in workergroup.go is not resolved, see that TODO */
+		// If the "active worker numbers" is like [2 2 ... 1 1], it means that an existing worker exited and the no new worker is started.
+		if slices.Equal([]int{1, 1}, activeNums[len(activeNums)-2:]) {
+			hasOnlyOneWorkerRunning.Store(true)
+		}
+		return nil
+	}
+	q, _ = newWorkerPoolQueueForTest("test-workpoolqueue", setting.QueueSettings{Type: "channel", BatchLength: 1, MaxWorkers: 2, Length: 100}, handler, false)
+	stop := runWorkerPoolQueue(q)
+	for i := range 100 {
+		assert.NoError(t, q.Push(i))
+	}
+	time.Sleep(500 * time.Millisecond)
+	assert.Greater(t, int(handledCount.Load()), 4) // make sure there are enough items handled during the test
+	assert.False(t, hasOnlyOneWorkerRunning.Load(), "a slow handler should not block other workers from starting")
+	stop()
 }