mirror of
https://github.com/hanzoai/forge.git
synced 2026-10-30 15:00:39 +00:00
zeekayandHanzo Dev
5014833cc1
The default queue was leveldb behind gitea.com/lunny/levelqueue — a second
embedded engine doing the job ZapDB already does for kms, cloud and mpc.
modules/queue/base_zapdb.go replaces it and gitea.com/lunny/levelqueue is out
of go.mod.
Layout is one prefix per queue so RemoveAll is a single DropPrefix:
<name>/i/<be-uint64> -> payload (big-endian, so key order IS FIFO order)
<name>/u/<payload> -> {} (unique index, only for unique queues)
Ordering comes from a ZapDB sequence rather than an in-memory counter, so a
restart mid-queue cannot hand out an index already on disk and overwrite an
unread item.
Two things this got wrong first, both caught by the suite:
PopItem must BLOCK until an item arrives or ctx ends — that is the contract
base_kv implements via backoffRetErr. Returning (nil, nil) on empty turned
the worker loop into a spin and made FlushWithContext report the cancellation
as a failure.
ZapDB takes an EXCLUSIVE directory lock, and every queue in the process —
issue indexer, code indexer, mail, webhook delivery, push update — shares one
DATADIR. Opening per queue therefore worked exactly once and then failed with
"Cannot acquire directory lock", surfacing far from the cause as a half-built
queue whose run loop nil-panicked. There is now one refcounted handle per
directory, which is what the leveldb backend got from the nosql manager and
is stated here rather than inherited.
"level" in an existing app.ini still starts — it resolves to zapdb.
TestManager fails under -count>1, and did so before this change too (verified in
a worktree at the previous commit, levelqueue present): the manager has no
unregister, so repeated runs in one process accumulate queues and FlushAll waits
on dead ones. -count=1 is green.
Everything else green; integration compile gate green.
Co-authored-by: Hanzo Dev <[email protected]>
118 lines
4.0 KiB
Go
118 lines
4.0 KiB
Go
// Copyright 2026 Hanzo AI, Inc. All rights reserved.
|
|
// Copyright 2023 The Gitea Authors. All rights reserved.
|
|
// SPDX-License-Identifier: MIT
|
|
|
|
package queue
|
|
|
|
import (
|
|
"path/filepath"
|
|
"testing"
|
|
|
|
"github.com/hanzoai/git/modules/setting"
|
|
|
|
"github.com/stretchr/testify/assert"
|
|
)
|
|
|
|
func TestManager(t *testing.T) {
|
|
setting.AppDataPath = t.TempDir()
|
|
|
|
newQueueFromConfig := func(name, cfg string) (*WorkerPoolQueue[int], error) {
|
|
cfgProvider, err := setting.NewConfigProviderFromData(cfg)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
qs, err := setting.GetQueueSettings(cfgProvider, name)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return newWorkerPoolQueueForTest(name, qs, func(s ...int) (unhandled []int) { return nil }, false)
|
|
}
|
|
|
|
// The old default parsed CONN_STR as a leveldb DSN and rejected a bad one
|
|
// here. The zapdb backend is addressed by DATADIR and does not read CONN_STR
|
|
// at all, so there is no DSN left to be invalid — CONN_STR now only means
|
|
// something to the kv backend, which validates it itself.
|
|
|
|
// test default config
|
|
q, err := newQueueFromConfig("default", "")
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, "default", q.GetName())
|
|
assert.Equal(t, "zapdb", q.GetType())
|
|
assert.Equal(t, filepath.Join(setting.AppDataPath, "queues/common"), q.baseConfig.DataFullDir)
|
|
assert.Equal(t, 100000, q.baseConfig.Length)
|
|
assert.Equal(t, 20, q.batchLength)
|
|
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())
|
|
assert.Equal(t, 0, q.GetWorkerNumber())
|
|
assert.Equal(t, 0, q.GetWorkerActiveNumber())
|
|
assert.Equal(t, 0, q.GetQueueItemNumber())
|
|
assert.Equal(t, "int", q.GetItemTypeName())
|
|
|
|
// test inherited config
|
|
cfgProvider, err := setting.NewConfigProviderFromData(`
|
|
[queue]
|
|
TYPE = channel
|
|
DATADIR = queues/dir1
|
|
LENGTH = 100
|
|
BATCH_LENGTH = 20
|
|
CONN_STR = "addrs=127.0.0.1:6379 db=0"
|
|
QUEUE_NAME = _queue1
|
|
|
|
[queue.sub]
|
|
TYPE = level
|
|
DATADIR = queues/dir2
|
|
LENGTH = 102
|
|
BATCH_LENGTH = 22
|
|
CONN_STR =
|
|
QUEUE_NAME = _q2
|
|
SET_NAME = _u2
|
|
MAX_WORKERS = 123
|
|
`)
|
|
|
|
assert.NoError(t, err)
|
|
|
|
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)
|
|
assert.Equal(t, 100, q1.baseConfig.Length)
|
|
assert.Equal(t, 20, q1.batchLength)
|
|
assert.Equal(t, "addrs=127.0.0.1:6379 db=0", q1.baseConfig.ConnStr)
|
|
assert.Equal(t, "no-such_queue1", q1.baseConfig.QueueFullName)
|
|
assert.Equal(t, "no-such_queue1_unique", q1.baseConfig.SetFullName)
|
|
assert.NotZero(t, q1.GetWorkerMaxNumber())
|
|
assert.Equal(t, 0, q1.GetWorkerNumber())
|
|
assert.Equal(t, 0, q1.GetWorkerActiveNumber())
|
|
assert.Equal(t, 0, q1.GetQueueItemNumber())
|
|
assert.Equal(t, "string", q1.GetItemTypeName())
|
|
qid1 := GetManager().qidCounter
|
|
|
|
q2 := createWorkerPoolQueue(t.Context(), "sub", cfgProvider, func(s ...int) (unhandled []int) { return nil }, false)
|
|
assert.Equal(t, "sub", q2.GetName())
|
|
assert.Equal(t, "zapdb", 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.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())
|
|
assert.Equal(t, 0, q2.GetWorkerNumber())
|
|
assert.Equal(t, 0, q2.GetWorkerActiveNumber())
|
|
assert.Equal(t, 0, q2.GetQueueItemNumber())
|
|
assert.Equal(t, "int", q2.GetItemTypeName())
|
|
qid2 := GetManager().qidCounter
|
|
|
|
assert.Equal(t, q1, GetManager().ManagedQueues()[qid1])
|
|
|
|
GetManager().GetManagedQueue(qid1).SetWorkerMaxNumber(120)
|
|
assert.Equal(t, 120, q1.workerMaxNum)
|
|
|
|
stop := runWorkerPoolQueue(q2)
|
|
assert.NoError(t, GetManager().GetManagedQueue(qid2).FlushWithContext(t.Context(), 0))
|
|
assert.NoError(t, GetManager().FlushAll(t.Context(), 0))
|
|
stop()
|
|
}
|