fix(queue): bump levelqueue to v0.5.0 and remove workarounds (#40219)

levelqueue v0.5.1 fixes this upstream
(https://gitea.com/lunny/levelqueue/pulls/8), so this now only bumps the
dependency.

A push wrote "high" before the payload, so an interrupted push left an
id counted with no key. `LPop` then returned `ErrNotFound` without
advancing "low" and the wrapper retried forever: nothing popped, no
worker started, later items unreachable. All were only fixable by
clearing the queue, losing the backlog:
- https://github.com/go-gitea/gitea/issues/32006 — mail, 0 workers
- https://github.com/go-gitea/gitea/issues/28919 — mirror, "already in
queue"
- https://github.com/go-gitea/gitea/issues/33862 — archives hanging
- https://github.com/go-gitea/gitea/issues/33882 — self-healing request

v0.5.1 writes counter and payload in one batch, moves the counters past
missing keys, and repairs leftover unique-queue set entries, so affected
instances recover on their next pop.

No test: the behaviour lives in the dependency, which covers it in its
own `recover_test.go`.

---------

Signed-off-by: wxiaoguang <wxiaoguang@gmail.com>
Co-authored-by: wxiaoguang <wxiaoguang@gmail.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

go.mod

@@ -7,7 +7,7 @@ toolchain go1.27.3
 require (
 	connectrpc.com/connect/v2 v2.0.2
 	gitea.com/lunny/dingtalk_webhook v0.0.0-20171025031554-e3534c89ef96
-	gitea.com/lunny/levelqueue v0.4.2-0.20230414023320-3c0159fe0fe4
+	gitea.com/lunny/levelqueue v0.5.1
 	gitea.dev/actionslib v1.5.2
 	gitea.dev/sdk v1.5.0
 	github.com/42wim/httpsig v1.3.0

go.sum

@@ -8,8 +8,8 @@ filippo.io/edwards25519 v1.2.0 h1:crnVqOiS4jqYleHd9vaKZ+HKtHfllngJIiOpNpoJsjo=
 filippo.io/edwards25519 v1.2.0/go.mod h1:xzAOLCNug/yB62zG1bQ8uziwrIqIuxhctzJT18Q77mc=
 gitea.com/lunny/dingtalk_webhook v0.0.0-20171025031554-e3534c89ef96 h1:+wWBi6Qfruqu7xJgjOIrKVQGiLUZdpKYCZewJ4clqhw=
 gitea.com/lunny/dingtalk_webhook v0.0.0-20171025031554-e3534c89ef96/go.mod h1:VyMQP6ue6MKHM8UsOXfNfuMKD0oSAWZdXVcpHIN2yaY=
-gitea.com/lunny/levelqueue v0.4.2-0.20230414023320-3c0159fe0fe4 h1:IFT+hup2xejHqdhS7keYWioqfmxdnfblFDTGoOwcZ+o=
-gitea.com/lunny/levelqueue v0.4.2-0.20230414023320-3c0159fe0fe4/go.mod h1:HBqmLbz56JWpfEGG0prskAV97ATNRoj5LDmPicD22hU=
+gitea.com/lunny/levelqueue v0.5.1 h1:15q1oBWaO2Aw9t3rDthQSTX+vBNgjYNkCgRJgES9gA8=
+gitea.com/lunny/levelqueue v0.5.1/go.mod h1:HBqmLbz56JWpfEGG0prskAV97ATNRoj5LDmPicD22hU=
 gitea.com/xorm/sqlfiddle v0.0.0-20180821085327-62ce714f951a h1:lSA0F4e9A2NcQSqGqTOXqu2aRi/XEQxDCBwM8yJtE6s=
 gitea.com/xorm/sqlfiddle v0.0.0-20180821085327-62ce714f951a/go.mod h1:EXuID2Zs0pAQhH8yz+DNjUbjppKQzKFAn28TMYPB6IU=
 gitea.dev/actionslib v1.5.2 h1:PSSUd+4jIjI+ycqo62+BRgAWkBnzjgeGQAcI3ZqH7LY=

modules/queue/base_levelqueue.go

@@ -5,10 +5,8 @@ package queue
 
 import (
 	"context"
-	"sync/atomic"
 
 	"gitea.dev/modules/nosql"
-	"gitea.dev/modules/queue/lqinternal"
 
 	"gitea.com/lunny/levelqueue"
 	"github.com/syndtr/goleveldb/leveldb"
@@ -16,7 +14,7 @@ import (
 
 type baseLevelQueue struct {
 	*baseLevelQueueCommonImpl
-	internal atomic.Pointer[levelqueue.Queue]
+	internal *levelqueue.Queue
 
 	conn string
 	cfg  *BaseConfig
@@ -41,12 +39,11 @@ func newBaseLevelQueueSimple(cfg *BaseConfig) (baseQueue, error) {
 		return nil, err
 	}
 	q := &baseLevelQueue{conn: conn, cfg: cfg, db: db}
-	lq, err := levelqueue.NewQueue(db, []byte(cfg.QueueFullName), false)
+	q.internal, err = levelqueue.NewQueue(db, []byte(cfg.QueueFullName), false)
 	if err != nil {
 		return nil, err
 	}
-	q.internal.Store(lq)
-	q.baseLevelQueueCommonImpl = baseLevelQueueCommon(q.cfg, nil, func() baseLevelQueuePushPoper { return q.internal.Load() })
+	q.baseLevelQueueCommonImpl = baseLevelQueueCommon(q.cfg, nil, q.internal)
 	return q, nil
 }
 
@@ -55,24 +52,16 @@ func (q *baseLevelQueue) HasItem(ctx context.Context, data []byte) (bool, error)
 }
 
 func (q *baseLevelQueue) Len(ctx context.Context) (int, error) {
-	return int(q.internal.Load().Len()), nil
+	return int(q.internal.Len()), nil
 }
 
 func (q *baseLevelQueue) Close() error {
-	err := q.internal.Load().Close()
+	err := q.internal.Close()
 	_ = nosql.GetManager().CloseLevelDB(q.conn)
 	q.db = nil // the db is not managed by us, it's managed by the nosql manager
 	return err
 }
 
 func (q *baseLevelQueue) RemoveAll(ctx context.Context) error {
-	lqinternal.RemoveLevelQueueKeys(q.db, []byte(q.cfg.QueueFullName))
-	lq, err := levelqueue.NewQueue(q.db, []byte(q.cfg.QueueFullName), false)
-	if err != nil {
-		return err
-	}
-	old := q.internal.Load()
-	q.internal.Store(lq)
-	_ = old.Close() // Not ideal for concurrency. Luckily, the levelqueue only sets its db=nil because it doesn't manage the db, so far so good
-	return nil
+	return q.internal.RemoveAll()
 }

modules/queue/base_levelqueue_common.go

@@ -26,9 +26,9 @@ type baseLevelQueuePushPoper interface {
 
 type baseLevelQueueCommonImpl struct {
 	*baseQueueNotifiable
-	length       int
-	internalFunc func() baseLevelQueuePushPoper
-	muCommon     *sync.Mutex
+	length   int
+	internal baseLevelQueuePushPoper
+	muCommon *sync.Mutex
 }
 
 func (q *baseLevelQueueCommonImpl) PushItem(ctx context.Context, data []byte) error {
@@ -38,11 +38,11 @@ func (q *baseLevelQueueCommonImpl) PushItem(ctx context.Context, data []byte) er
 			defer q.muCommon.Unlock()
 		}
 
-		cnt := int(q.internalFunc().Len())
+		cnt := int(q.internal.Len())
 		if cnt >= q.length {
 			return true, nil, nil
 		}
-		retry, err = false, q.internalFunc().RPush(data)
+		retry, err = false, q.internal.RPush(data)
 		if err == levelqueue.ErrAlreadyInQueue {
 			err = ErrAlreadyInQueue
 		}
@@ -61,7 +61,7 @@ func (q *baseLevelQueueCommonImpl) PopItem(ctx context.Context) ([]byte, error)
 			defer q.muCommon.Unlock()
 		}
 
-		data, err = q.internalFunc().LPop()
+		data, err = q.internal.LPop()
 		if err == levelqueue.ErrNotFound {
 			return true, nil, nil
 		}
@@ -72,8 +72,8 @@ func (q *baseLevelQueueCommonImpl) PopItem(ctx context.Context) ([]byte, error)
 	})
 }
 
-func baseLevelQueueCommon(cfg *BaseConfig, mu *sync.Mutex, internalFunc func() baseLevelQueuePushPoper) *baseLevelQueueCommonImpl {
-	return &baseLevelQueueCommonImpl{length: cfg.Length, muCommon: mu, internalFunc: internalFunc, baseQueueNotifiable: newBaseQueueNotifiable()}
+func baseLevelQueueCommon(cfg *BaseConfig, mu *sync.Mutex, internal baseLevelQueuePushPoper) *baseLevelQueueCommonImpl {
+	return &baseLevelQueueCommonImpl{length: cfg.Length, muCommon: mu, internal: internal, baseQueueNotifiable: newBaseQueueNotifiable()}
 }
 
 func prepareLevelDB(cfg *BaseConfig) (conn string, db *leveldb.DB, err error) {

modules/queue/base_levelqueue_test.go

@@ -6,13 +6,9 @@ package queue
 import (
 	"testing"
 
-	"gitea.dev/modules/queue/lqinternal"
 	"gitea.dev/modules/setting"
 
-	"gitea.com/lunny/levelqueue"
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-	"github.com/syndtr/goleveldb/leveldb"
 )
 
 func TestBaseLevelDB(t *testing.T) {
@@ -27,53 +23,3 @@ func TestBaseLevelDB(t *testing.T) {
 	testQueueBasic(t, newBaseLevelQueueSimple, toBaseConfig("baseLevelQueue", setting.QueueSettings{Datadir: t.TempDir() + "/queue-test", Length: 10}), optsSimple)
 	testQueueBasic(t, newBaseLevelQueueUnique, toBaseConfig("baseLevelQueueUnique", setting.QueueSettings{ConnStr: "leveldb://" + t.TempDir() + "/queue-test", Length: 10}), optsUnique)
 }
-
-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)
-	require.NoError(t, err)
-	defer db.Close()
-
-	assert.NoError(t, db.Put([]byte("other-key"), []byte("other-value"), nil))
-
-	nameQueuePrefix := []byte("queue_name")
-	nameSetPrefix := []byte("set_name")
-	lq, err := levelqueue.NewUniqueQueue(db, nameQueuePrefix, nameSetPrefix, false)
-	assert.NoError(t, err)
-	assert.NoError(t, lq.RPush([]byte("item-1")))
-
-	itemKey := lqinternal.QueueItemKeyBytes(nameQueuePrefix, 1)
-	itemValue, err := db.Get(itemKey, nil)
-	assert.NoError(t, err)
-	assert.Equal(t, []byte("item-1"), itemValue)
-
-	// there should be 5 keys in db: queue low, queue high, 1 queue item, 1 set item, and "other-key"
-	keys := lqinternal.ListLevelQueueKeys(db)
-	assert.Len(t, keys, 5)
-
-	// delete the queue item key, to corrupt the queue
-	assert.NoError(t, db.Delete(itemKey, nil))
-	// now the queue is corrupted, it never works again
-	_, err = lq.LPop()
-	assert.ErrorIs(t, err, levelqueue.ErrNotFound)
-	assert.NoError(t, lq.Close())
-
-	// remove all the queue related keys to reset the queue
-	lqinternal.RemoveLevelQueueKeys(db, nameQueuePrefix)
-	lqinternal.RemoveLevelQueueKeys(db, nameSetPrefix)
-	// now there should be only 1 key in db: "other-key"
-	keys = lqinternal.ListLevelQueueKeys(db)
-	assert.Len(t, keys, 1)
-	assert.Equal(t, []byte("other-key"), keys[0])
-
-	// re-create a queue from db
-	lq, err = levelqueue.NewUniqueQueue(db, nameQueuePrefix, nameSetPrefix, false)
-	assert.NoError(t, err)
-	assert.NoError(t, lq.RPush([]byte("item-new-1")))
-	// now the queue works again
-	itemValue, err = lq.LPop()
-	assert.NoError(t, err)
-	assert.Equal(t, []byte("item-new-1"), itemValue)
-	assert.NoError(t, lq.Close())
-}

modules/queue/base_levelqueue_unique.go

@@ -6,10 +6,8 @@ package queue
 import (
 	"context"
 	"sync"
-	"sync/atomic"
 
 	"gitea.dev/modules/nosql"
-	"gitea.dev/modules/queue/lqinternal"
 
 	"gitea.com/lunny/levelqueue"
 	"github.com/syndtr/goleveldb/leveldb"
@@ -17,13 +15,13 @@ import (
 
 type baseLevelQueueUnique struct {
 	*baseLevelQueueCommonImpl
-	internal atomic.Pointer[levelqueue.UniqueQueue]
+	internal *levelqueue.UniqueQueue
 
 	conn string
 	cfg  *BaseConfig
 	db   *leveldb.DB
 
-	muBase sync.Mutex // the levelqueue.UniqueQueue is not thread-safe, there is no mutex protecting the underlying queue&set together
+	muBase sync.Mutex // protects multi-step queue+set operations, e.g. the length check followed by a push
 }
 
 var (
@@ -37,31 +35,30 @@ func newBaseLevelQueueUnique(cfg *BaseConfig) (baseQueue, error) {
 		return nil, err
 	}
 	q := &baseLevelQueueUnique{conn: conn, cfg: cfg, db: db}
-	lq, err := levelqueue.NewUniqueQueue(db, []byte(cfg.QueueFullName), []byte(cfg.SetFullName), false)
+	q.internal, err = levelqueue.NewUniqueQueue(db, []byte(cfg.QueueFullName), []byte(cfg.SetFullName), false)
 	if err != nil {
 		return nil, err
 	}
-	q.internal.Store(lq)
-	q.baseLevelQueueCommonImpl = baseLevelQueueCommon(q.cfg, &q.muBase, func() baseLevelQueuePushPoper { return q.internal.Load() })
+	q.baseLevelQueueCommonImpl = baseLevelQueueCommon(q.cfg, &q.muBase, q.internal)
 	return q, nil
 }
 
 func (q *baseLevelQueueUnique) HasItem(ctx context.Context, data []byte) (bool, error) {
 	q.muBase.Lock()
 	defer q.muBase.Unlock()
-	return q.internal.Load().Has(data)
+	return q.internal.Has(data)
 }
 
 func (q *baseLevelQueueUnique) Len(ctx context.Context) (int, error) {
 	q.muBase.Lock()
 	defer q.muBase.Unlock()
-	return int(q.internal.Load().Len()), nil
+	return int(q.internal.Len()), nil
 }
 
 func (q *baseLevelQueueUnique) Close() error {
 	q.muBase.Lock()
 	defer q.muBase.Unlock()
-	err := q.internal.Load().Close()
+	err := q.internal.Close()
 	q.db = nil // the db is not managed by us, it's managed by the nosql manager
 	_ = nosql.GetManager().CloseLevelDB(q.conn)
 	return err
@@ -70,14 +67,5 @@ func (q *baseLevelQueueUnique) Close() error {
 func (q *baseLevelQueueUnique) RemoveAll(ctx context.Context) error {
 	q.muBase.Lock()
 	defer q.muBase.Unlock()
-	lqinternal.RemoveLevelQueueKeys(q.db, []byte(q.cfg.QueueFullName))
-	lqinternal.RemoveLevelQueueKeys(q.db, []byte(q.cfg.SetFullName))
-	lq, err := levelqueue.NewUniqueQueue(q.db, []byte(q.cfg.QueueFullName), []byte(q.cfg.SetFullName), false)
-	if err != nil {
-		return err
-	}
-	old := q.internal.Load()
-	q.internal.Store(lq)
-	_ = old.Close() // Not ideal for concurrency. Luckily, the levelqueue only sets its db=nil because it doesn't manage the db, so far so good
-	return nil
+	return q.internal.RemoveAll()
 }

modules/queue/base_test.go

@@ -25,7 +25,7 @@ func testQueueBasic(t *testing.T, newFn func(cfg *BaseConfig) (baseQueue, error)
 		assert.NoError(t, err)
 
 		ctx := t.Context()
-		_ = q.RemoveAll(ctx)
+		assert.NoError(t, q.RemoveAll(ctx))
 		cnt, err := q.Len(ctx)
 		assert.NoError(t, err)
 		assert.Equal(t, 0, cnt)
@@ -138,11 +138,14 @@ func testQueueBasic(t *testing.T, newFn func(cfg *BaseConfig) (baseQueue, error)
 		assert.NoError(t, err)
 		assert.Equal(t, cfg.Length, cnt)
 
-		_ = q.RemoveAll(ctx)
+		assert.NoError(t, q.RemoveAll(ctx))
 
 		cnt, err = q.Len(ctx)
 		assert.NoError(t, err)
 		assert.Equal(t, 0, cnt)
+
+		// a removed payload must not stay reserved in the set, or this returns ErrAlreadyInQueue
+		assert.NoError(t, q.PushItem(ctx, []byte("item-0")))
 	})
 }
 

modules/queue/lqinternal/lqinternal.go

@@ -1,48 +0,0 @@
-// Copyright 2023 The Gitea Authors. All rights reserved.
-// SPDX-License-Identifier: MIT
-
-package lqinternal
-
-import (
-	"bytes"
-	"encoding/binary"
-
-	"github.com/syndtr/goleveldb/leveldb"
-	"github.com/syndtr/goleveldb/leveldb/opt"
-)
-
-func QueueItemIDBytes(id int64) []byte {
-	buf := make([]byte, 8)
-	binary.PutVarint(buf, id)
-	return buf
-}
-
-func QueueItemKeyBytes(prefix []byte, id int64) []byte {
-	key := make([]byte, len(prefix), len(prefix)+1+8)
-	copy(key, prefix)
-	key = append(key, '-')
-	return append(key, QueueItemIDBytes(id)...)
-}
-
-func RemoveLevelQueueKeys(db *leveldb.DB, namePrefix []byte) {
-	keyPrefix := make([]byte, len(namePrefix)+1)
-	copy(keyPrefix, namePrefix)
-	keyPrefix[len(namePrefix)] = '-'
-
-	it := db.NewIterator(nil, &opt.ReadOptions{Strict: opt.NoStrict})
-	defer it.Release()
-	for it.Next() {
-		if bytes.HasPrefix(it.Key(), keyPrefix) {
-			_ = db.Delete(it.Key(), nil)
-		}
-	}
-}
-
-func ListLevelQueueKeys(db *leveldb.DB) (res [][]byte) {
-	it := db.NewIterator(nil, &opt.ReadOptions{Strict: opt.NoStrict})
-	defer it.Release()
-	for it.Next() {
-		res = append(res, it.Key())
-	}
-	return res
-}