mirror of
https://github.com/go-gitea/gitea.git
synced 2026-11-09 17:37:20 -06:00
87a703819b
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>
72 lines
1.7 KiB
Go
72 lines
1.7 KiB
Go
// Copyright 2023 The Gitea Authors. All rights reserved.
|
|
// SPDX-License-Identifier: MIT
|
|
|
|
package queue
|
|
|
|
import (
|
|
"context"
|
|
"sync"
|
|
|
|
"gitea.dev/modules/nosql"
|
|
|
|
"gitea.com/lunny/levelqueue"
|
|
"github.com/syndtr/goleveldb/leveldb"
|
|
)
|
|
|
|
type baseLevelQueueUnique struct {
|
|
*baseLevelQueueCommonImpl
|
|
internal *levelqueue.UniqueQueue
|
|
|
|
conn string
|
|
cfg *BaseConfig
|
|
db *leveldb.DB
|
|
|
|
muBase sync.Mutex // protects multi-step queue+set operations, e.g. the length check followed by a push
|
|
}
|
|
|
|
var (
|
|
_ baseQueue = (*baseLevelQueueUnique)(nil)
|
|
_ baseQueueNotifiableInterface = (*baseLevelQueueUnique)(nil)
|
|
)
|
|
|
|
func newBaseLevelQueueUnique(cfg *BaseConfig) (baseQueue, error) {
|
|
conn, db, err := prepareLevelDB(cfg)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
q := &baseLevelQueueUnique{conn: conn, cfg: cfg, db: db}
|
|
q.internal, err = levelqueue.NewUniqueQueue(db, []byte(cfg.QueueFullName), []byte(cfg.SetFullName), false)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
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.Has(data)
|
|
}
|
|
|
|
func (q *baseLevelQueueUnique) Len(ctx context.Context) (int, error) {
|
|
q.muBase.Lock()
|
|
defer q.muBase.Unlock()
|
|
return int(q.internal.Len()), nil
|
|
}
|
|
|
|
func (q *baseLevelQueueUnique) Close() error {
|
|
q.muBase.Lock()
|
|
defer q.muBase.Unlock()
|
|
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
|
|
}
|
|
|
|
func (q *baseLevelQueueUnique) RemoveAll(ctx context.Context) error {
|
|
q.muBase.Lock()
|
|
defer q.muBase.Unlock()
|
|
return q.internal.RemoveAll()
|
|
}
|