nik/navidrome
1
0
Fork 0
mirror of https://github.com/navidrome/navidrome.git synced 2026-08-31 07:30:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(plugins): increase maxConcurrency for task queue and handle budget exhaustion

Browse Source 
Signed-off-by: Deluan <deluan@navidrome.org>
This commit is contained in:
Deluan 2026-02-26 20:40:37 -05:00 committed by Deluan Quintão
parent 2a0573fa51
commit 258fabf2f5
3 changed files with 30 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
plugins/host_taskqueue.go
View File

@ -175,18 +175,25 @@ func (s *taskQueueServiceImpl) applyConfigDefaults(ctx context.Context, name str
}
// clampConcurrency reduces config.Concurrency if it exceeds the remaining budget.
// Returns an error when the concurrency budget is fully exhausted.
// Must be called with s.mu held.
func (s *taskQueueServiceImpl) clampConcurrency(ctx context.Context, name string, config *host.QueueConfig) {
func (s *taskQueueServiceImpl) clampConcurrency(ctx context.Context, name string, config *host.QueueConfig) error {
var allocated int32
for _, qs := range s.queues {
allocated += qs.config.Concurrency
}
available := max(s.maxConcurrency-allocated, 1)
available := s.maxConcurrency - allocated
if available <= 0 {
log.Warn(ctx, "TaskQueue concurrency budget exhausted", "plugin", s.pluginName, "queue", name,
"allocated", allocated, "maxConcurrency", s.maxConcurrency)
return fmt.Errorf("concurrency budget exhausted (%d/%d allocated)", allocated, s.maxConcurrency)
}
if config.Concurrency > available {
log.Warn(ctx, "TaskQueue concurrency clamped", "plugin", s.pluginName, "queue", name,
"requested", config.Concurrency, "available", available, "maxConcurrency", s.maxConcurrency)
config.Concurrency = available
}
return nil
}
func (s *taskQueueServiceImpl) CreateQueue(ctx context.Context, name string, config host.QueueConfig) error {
@ -202,7 +209,9 @@ func (s *taskQueueServiceImpl) CreateQueue(ctx context.Context, name string, con
s.mu.Lock()
defer s.mu.Unlock()
s.clampConcurrency(ctx, name, &config)
if err := s.clampConcurrency(ctx, name, &config); err != nil {
return err
}
if _, exists := s.queues[name]; exists {
return fmt.Errorf("queue %q already exists", name)

19
plugins/host_taskqueue_test.go
View File

@ -157,7 +157,22 @@ var _ = Describe("TaskQueueService", func() {
service.mu.Lock()
qs := service.queues["clamped-queue"]
service.mu.Unlock()
Expect(qs.config.Concurrency).To(BeNumerically("<=", int32(5)))
Expect(qs.config.Concurrency).To(Equal(int32(5)))
})
It("returns error when concurrency budget is exhausted", func() {
// maxConcurrency is 5; create a queue that uses all 5
err := service.CreateQueue(ctx, "full-budget", host.QueueConfig{
Concurrency: 5,
})
Expect(err).ToNot(HaveOccurred())
// Next queue should fail — no budget remaining
err = service.CreateQueue(ctx, "over-budget", host.QueueConfig{
Concurrency: 1,
})
Expect(err).To(HaveOccurred())
Expect(err.Error()).To(ContainSubstring("concurrency budget exhausted"))
})
It("clamps retention below minimum", func() {
@ -727,7 +742,7 @@ var _ = Describe("TaskQueueService Integration", Ordered, func() {
Expect(ok).To(BeTrue())
Expect(p.manifest.Permissions).ToNot(BeNil())
Expect(p.manifest.Permissions.Taskqueue).ToNot(BeNil())
Expect(p.manifest.Permissions.Taskqueue.MaxConcurrency).To(Equal(3))
Expect(p.manifest.Permissions.Taskqueue.MaxConcurrency).To(Equal(10))
Expect(p.capabilities).To(ContainElement(CapabilityTaskWorker))
})
})

2
plugins/testdata/test-taskqueue/manifest.json vendored
View File

@ -6,7 +6,7 @@
"permissions": {
"taskqueue": {
"reason": "For testing task queue operations",
"maxConcurrency": 3
"maxConcurrency": 10
}
}
}