diff --git a/go.mod b/go.mod index 53c7418c810..1d7f26bdb1a 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index ce8c9ce1e00..df7203778a7 100644 --- a/go.sum +++ b/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= diff --git a/modules/queue/base_levelqueue.go b/modules/queue/base_levelqueue.go index 24f382913a0..caa02c60238 100644 --- a/modules/queue/base_levelqueue.go +++ b/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() } diff --git a/modules/queue/base_levelqueue_common.go b/modules/queue/base_levelqueue_common.go index 9d5e578d8c3..55238e0f4ba 100644 --- a/modules/queue/base_levelqueue_common.go +++ b/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) { diff --git a/modules/queue/base_levelqueue_test.go b/modules/queue/base_levelqueue_test.go index 7c4166bc5e3..163d5fbe0c2 100644 --- a/modules/queue/base_levelqueue_test.go +++ b/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()) -} diff --git a/modules/queue/base_levelqueue_unique.go b/modules/queue/base_levelqueue_unique.go index 60de3785a15..33d37d8e0b8 100644 --- a/modules/queue/base_levelqueue_unique.go +++ b/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() } diff --git a/modules/queue/base_test.go b/modules/queue/base_test.go index bc9e6167a99..32bd5ab0624 100644 --- a/modules/queue/base_test.go +++ b/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"))) }) } diff --git a/modules/queue/lqinternal/lqinternal.go b/modules/queue/lqinternal/lqinternal.go deleted file mode 100644 index 89aa4e59894..00000000000 --- a/modules/queue/lqinternal/lqinternal.go +++ /dev/null @@ -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 -}