From 258fabf2f58565720996b870b0f0beaeac603d89 Mon Sep 17 00:00:00 2001
From: Deluan <deluan@navidrome.org>
Date: Thu, 26 Feb 2026 20:40:37 -0500
Subject: [PATCH] feat(plugins): increase maxConcurrency for task queue and
 handle budget exhaustion

Signed-off-by: Deluan <deluan@navidrome.org>
---
 plugins/host_taskqueue.go                     | 15 ++++++++++++---
 plugins/host_taskqueue_test.go                | 19 +++++++++++++++++--
 plugins/testdata/test-taskqueue/manifest.json |  2 +-
 3 files changed, 30 insertions(+), 6 deletions(-)

diff --git a/plugins/host_taskqueue.go b/plugins/host_taskqueue.go
index 336ae1fe6..a85c66d4f 100644
--- a/plugins/host_taskqueue.go
+++ b/plugins/host_taskqueue.go
@@ -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)
diff --git a/plugins/host_taskqueue_test.go b/plugins/host_taskqueue_test.go
index d13dc70a5..a66c25085 100644
--- a/plugins/host_taskqueue_test.go
+++ b/plugins/host_taskqueue_test.go
@@ -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))
 		})
 	})
diff --git a/plugins/testdata/test-taskqueue/manifest.json b/plugins/testdata/test-taskqueue/manifest.json
index 2e3695628..3cd3b0f0b 100644
--- a/plugins/testdata/test-taskqueue/manifest.json
+++ b/plugins/testdata/test-taskqueue/manifest.json
@@ -6,7 +6,7 @@
   "permissions": {
     "taskqueue": {
       "reason": "For testing task queue operations",
-      "maxConcurrency": 3
+      "maxConcurrency": 10
     }
   }
 }