feat(plugins): implement TaskQueue service with SQLite persistence and workers

Per-plugin SQLite database with queues and tasks tables. Worker goroutines dequeue tasks and invoke nd_task_execute callback. Exponential backoff retries, rate limiting via delayMs, automatic cleanup of terminal tasks.
2026-08-31 07:30:32 +00:00 · 2026-02-26 15:27:57 -05:00 · 2026-02-26 15:27:57 -05:00 · fa27138ab3
commit fa27138ab3
parent 2742453e60
2 changed files with 1039 additions and 0 deletions
--- a/plugins/host_taskqueue.go
+++ b/plugins/host_taskqueue.go
@ -0,0 +1,557 @@
+package plugins
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"io"
+	"math"
+	"os"
+	"path/filepath"
+	"sync"
+	"time"
+
+	_ "github.com/mattn/go-sqlite3"
+	"github.com/navidrome/navidrome/conf"
+	"github.com/navidrome/navidrome/log"
+	"github.com/navidrome/navidrome/model/id"
+	"github.com/navidrome/navidrome/plugins/capabilities"
+	"github.com/navidrome/navidrome/plugins/host"
+)
+
+const (
+	defaultConcurrency int32 = 1
+	defaultBackoffMs   int64 = 1000
+	defaultRetentionMs int64 = 3_600_000   // 1 hour
+	minRetentionMs     int64 = 60_000      // 1 minute
+	maxRetentionMs     int64 = 604_800_000 // 1 week
+	cleanupInterval          = 5 * time.Minute
+	pollInterval             = 5 * time.Second
+	shutdownTimeout          = 10 * time.Second
+)
+
+// CapabilityTaskWorker indicates the plugin can receive task execution callbacks.
+// Detected when the plugin exports the task worker callback function.
+const CapabilityTaskWorker Capability = "TaskWorker"
+
+const FuncTaskWorkerCallback = "nd_task_execute"
+
+func init() {
+	registerCapability(CapabilityTaskWorker, FuncTaskWorkerCallback)
+}
+
+// queueState holds in-memory state for a single task queue.
+type queueState struct {
+	config         host.QueueConfig
+	signal         chan struct{}
+	lastDispatchAt time.Time
+	mu             sync.Mutex
+}
+
+// taskQueueServiceImpl implements host.TaskQueueService with SQLite persistence
+// and background worker goroutines for task execution.
+type taskQueueServiceImpl struct {
+	pluginName     string
+	manager        *Manager
+	maxConcurrency int32
+	db             *sql.DB
+	ctx            context.Context
+	cancel         context.CancelFunc
+	wg             sync.WaitGroup
+	mu             sync.Mutex
+	queues         map[string]*queueState
+
+	// For testing: override how callbacks are invoked
+	invokeCallbackFn func(ctx context.Context, queueName, taskID string, payload []byte, attempt int32) error
+}
+
+// newTaskQueueService creates a new taskQueueServiceImpl with its own SQLite database.
+func newTaskQueueService(pluginName string, manager *Manager, maxConcurrency int32) (*taskQueueServiceImpl, error) {
+	// Create plugin data directory
+	dataDir := filepath.Join(conf.Server.DataFolder, "plugins", pluginName)
+	if err := os.MkdirAll(dataDir, 0700); err != nil {
+		return nil, fmt.Errorf("creating plugin data directory: %w", err)
+	}
+
+	// Open SQLite database
+	dbPath := filepath.Join(dataDir, "taskqueue.db")
+	db, err := sql.Open("sqlite3", dbPath+"?_busy_timeout=5000&_journal_mode=WAL&_foreign_keys=off")
+	if err != nil {
+		return nil, fmt.Errorf("opening taskqueue database: %w", err)
+	}
+
+	db.SetMaxOpenConns(3)
+	db.SetMaxIdleConns(1)
+
+	// Create schema
+	if err := createTaskQueueSchema(db); err != nil {
+		db.Close()
+		return nil, fmt.Errorf("creating taskqueue schema: %w", err)
+	}
+
+	ctx, cancel := context.WithCancel(manager.ctx)
+
+	s := &taskQueueServiceImpl{
+		pluginName:     pluginName,
+		manager:        manager,
+		maxConcurrency: maxConcurrency,
+		db:             db,
+		ctx:            ctx,
+		cancel:         cancel,
+		queues:         make(map[string]*queueState),
+	}
+	s.invokeCallbackFn = s.defaultInvokeCallback
+
+	// Start cleanup goroutine
+	s.wg.Add(1)
+	go s.cleanupLoop()
+
+	log.Debug("Initialized plugin taskqueue", "plugin", pluginName, "path", dbPath, "maxConcurrency", maxConcurrency)
+	return s, nil
+}
+
+func createTaskQueueSchema(db *sql.DB) error {
+	_, err := db.Exec(`
+		CREATE TABLE IF NOT EXISTS queues (
+			name TEXT PRIMARY KEY,
+			concurrency INTEGER NOT NULL DEFAULT 1,
+			max_retries INTEGER NOT NULL DEFAULT 0,
+			backoff_ms INTEGER NOT NULL DEFAULT 1000,
+			delay_ms INTEGER NOT NULL DEFAULT 0,
+			retention_ms INTEGER NOT NULL DEFAULT 3600000
+		);
+
+		CREATE TABLE IF NOT EXISTS tasks (
+			id TEXT PRIMARY KEY,
+			queue_name TEXT NOT NULL REFERENCES queues(name),
+			payload BLOB NOT NULL,
+			status TEXT NOT NULL DEFAULT 'pending',
+			attempt INTEGER NOT NULL DEFAULT 0,
+			max_retries INTEGER NOT NULL,
+			next_run_at INTEGER NOT NULL,
+			created_at INTEGER NOT NULL,
+			updated_at INTEGER NOT NULL
+		);
+
+		CREATE INDEX IF NOT EXISTS idx_tasks_dequeue ON tasks(queue_name, status, next_run_at);
+	`)
+	return err
+}
+
+// CreateQueue creates a named task queue with the given configuration.
+func (s *taskQueueServiceImpl) CreateQueue(ctx context.Context, name string, config host.QueueConfig) error {
+	// Apply defaults
+	if config.Concurrency <= 0 {
+		config.Concurrency = defaultConcurrency
+	}
+	if config.BackoffMs <= 0 {
+		config.BackoffMs = defaultBackoffMs
+	}
+	if config.RetentionMs == 0 {
+		config.RetentionMs = defaultRetentionMs
+	}
+
+	// Clamp retention
+	if config.RetentionMs < minRetentionMs {
+		log.Warn(ctx, "TaskQueue retention clamped to minimum", "plugin", s.pluginName, "queue", name,
+			"requested", config.RetentionMs, "min", minRetentionMs)
+		config.RetentionMs = minRetentionMs
+	}
+	if config.RetentionMs > maxRetentionMs {
+		log.Warn(ctx, "TaskQueue retention clamped to maximum", "plugin", s.pluginName, "queue", name,
+			"requested", config.RetentionMs, "max", maxRetentionMs)
+		config.RetentionMs = maxRetentionMs
+	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	// Clamp concurrency based on maxConcurrency minus already-allocated concurrency
+	var allocated int32
+	for _, qs := range s.queues {
+		allocated += qs.config.Concurrency
+	}
+	available := s.maxConcurrency - allocated
+	if available <= 0 {
+		available = 1 // Always allow at least 1
+	}
+	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
+	}
+
+	// Check queue name doesn't already exist
+	if _, exists := s.queues[name]; exists {
+		return fmt.Errorf("queue %q already exists", name)
+	}
+
+	// Insert into queues table
+	_, err := s.db.ExecContext(ctx, `
+		INSERT INTO queues (name, concurrency, max_retries, backoff_ms, delay_ms, retention_ms)
+		VALUES (?, ?, ?, ?, ?, ?)
+	`, name, config.Concurrency, config.MaxRetries, config.BackoffMs, config.DelayMs, config.RetentionMs)
+	if err != nil {
+		return fmt.Errorf("creating queue: %w", err)
+	}
+
+	// Reset stale running tasks from previous crash
+	now := time.Now().UnixMilli()
+	_, err = s.db.ExecContext(ctx, `
+		UPDATE tasks SET status = 'pending', updated_at = ? WHERE queue_name = ? AND status = 'running'
+	`, now, name)
+	if err != nil {
+		return fmt.Errorf("resetting stale tasks: %w", err)
+	}
+
+	// Store queue state
+	qs := &queueState{
+		config: config,
+		signal: make(chan struct{}, 1),
+	}
+	s.queues[name] = qs
+
+	// Start worker goroutines
+	for i := int32(0); i < config.Concurrency; i++ {
+		s.wg.Add(1)
+		go s.worker(name, qs)
+	}
+
+	log.Debug(ctx, "Created task queue", "plugin", s.pluginName, "queue", name,
+		"concurrency", config.Concurrency, "maxRetries", config.MaxRetries,
+		"backoffMs", config.BackoffMs, "delayMs", config.DelayMs, "retentionMs", config.RetentionMs)
+	return nil
+}
+
+// Enqueue adds a task to the named queue and returns the task ID.
+func (s *taskQueueServiceImpl) Enqueue(ctx context.Context, queueName string, payload []byte) (string, error) {
+	s.mu.Lock()
+	qs, exists := s.queues[queueName]
+	s.mu.Unlock()
+
+	if !exists {
+		return "", fmt.Errorf("queue %q does not exist", queueName)
+	}
+
+	taskID := id.NewRandom()
+	now := time.Now().UnixMilli()
+
+	_, err := s.db.ExecContext(ctx, `
+		INSERT INTO tasks (id, queue_name, payload, status, attempt, max_retries, next_run_at, created_at, updated_at)
+		VALUES (?, ?, ?, 'pending', 0, ?, ?, ?, ?)
+	`, taskID, queueName, payload, qs.config.MaxRetries, now, now, now)
+	if err != nil {
+		return "", fmt.Errorf("enqueuing task: %w", err)
+	}
+
+	// Signal workers (non-blocking)
+	select {
+	case qs.signal <- struct{}{}:
+	default:
+	}
+
+	log.Trace(ctx, "Enqueued task", "plugin", s.pluginName, "queue", queueName, "taskID", taskID)
+	return taskID, nil
+}
+
+// GetTaskStatus returns the status of a task.
+func (s *taskQueueServiceImpl) GetTaskStatus(ctx context.Context, taskID string) (string, error) {
+	var status string
+	err := s.db.QueryRowContext(ctx, `SELECT status FROM tasks WHERE id = ?`, taskID).Scan(&status)
+	if errors.Is(err, sql.ErrNoRows) {
+		return "", fmt.Errorf("task %q not found", taskID)
+	}
+	if err != nil {
+		return "", fmt.Errorf("getting task status: %w", err)
+	}
+	return status, nil
+}
+
+// CancelTask cancels a pending task.
+func (s *taskQueueServiceImpl) CancelTask(ctx context.Context, taskID string) error {
+	now := time.Now().UnixMilli()
+	result, err := s.db.ExecContext(ctx, `
+		UPDATE tasks SET status = 'cancelled', updated_at = ? WHERE id = ? AND status = 'pending'
+	`, now, taskID)
+	if err != nil {
+		return fmt.Errorf("cancelling task: %w", err)
+	}
+
+	rowsAffected, err := result.RowsAffected()
+	if err != nil {
+		return fmt.Errorf("checking cancel result: %w", err)
+	}
+
+	if rowsAffected == 0 {
+		// Check if task exists at all
+		var status string
+		err := s.db.QueryRowContext(ctx, `SELECT status FROM tasks WHERE id = ?`, taskID).Scan(&status)
+		if errors.Is(err, sql.ErrNoRows) {
+			return fmt.Errorf("task %q not found", taskID)
+		}
+		if err != nil {
+			return fmt.Errorf("checking task existence: %w", err)
+		}
+		return fmt.Errorf("task %q cannot be cancelled (status: %s)", taskID, status)
+	}
+
+	log.Trace(ctx, "Cancelled task", "plugin", s.pluginName, "taskID", taskID)
+	return nil
+}
+
+// worker is the main loop for a single worker goroutine.
+func (s *taskQueueServiceImpl) worker(queueName string, qs *queueState) {
+	defer s.wg.Done()
+
+	// Process any existing pending tasks immediately on startup
+	s.drainQueue(queueName, qs)
+
+	ticker := time.NewTicker(pollInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-s.ctx.Done():
+			return
+		case <-qs.signal:
+			s.drainQueue(queueName, qs)
+		case <-ticker.C:
+			s.drainQueue(queueName, qs)
+		}
+	}
+}
+
+// drainQueue processes tasks until the queue is empty.
+func (s *taskQueueServiceImpl) drainQueue(queueName string, qs *queueState) {
+	for {
+		if s.ctx.Err() != nil {
+			return
+		}
+		if !s.processTask(queueName, qs) {
+			return
+		}
+	}
+}
+
+// processTask dequeues and processes a single task. Returns true if a task was processed.
+func (s *taskQueueServiceImpl) processTask(queueName string, qs *queueState) bool {
+	now := time.Now().UnixMilli()
+
+	// Atomically dequeue a task
+	var taskID string
+	var payload []byte
+	var attempt int32
+	var maxRetries int32
+	err := s.db.QueryRowContext(s.ctx, `
+		UPDATE tasks SET status = 'running', attempt = attempt + 1, updated_at = ?
+		WHERE id = (
+			SELECT id FROM tasks
+			WHERE queue_name = ? AND status = 'pending' AND next_run_at <= ?
+			ORDER BY next_run_at, created_at LIMIT 1
+		)
+		RETURNING id, payload, attempt, max_retries
+	`, now, queueName, now).Scan(&taskID, &payload, &attempt, &maxRetries)
+	if errors.Is(err, sql.ErrNoRows) {
+		return false
+	}
+	if err != nil {
+		log.Error(s.ctx, "Failed to dequeue task", "plugin", s.pluginName, "queue", queueName, err)
+		return false
+	}
+
+	// Enforce delay between task dispatches
+	if qs.config.DelayMs > 0 {
+		qs.mu.Lock()
+		elapsed := time.Since(qs.lastDispatchAt)
+		delay := time.Duration(qs.config.DelayMs) * time.Millisecond
+		if elapsed < delay {
+			waitTime := delay - elapsed
+			qs.mu.Unlock()
+
+			select {
+			case <-s.ctx.Done():
+				// Put the task back to pending on shutdown
+				s.revertTaskToPending(taskID)
+				return false
+			case <-time.After(waitTime):
+			}
+
+			qs.mu.Lock()
+		}
+		qs.lastDispatchAt = time.Now()
+		qs.mu.Unlock()
+	}
+
+	// Invoke callback
+	log.Debug(s.ctx, "Executing task", "plugin", s.pluginName, "queue", queueName, "taskID", taskID, "attempt", attempt)
+	callbackErr := s.invokeCallbackFn(s.ctx, queueName, taskID, payload, attempt)
+
+	// If context was cancelled (shutdown), revert task to pending for recovery
+	if s.ctx.Err() != nil {
+		s.revertTaskToPending(taskID)
+		return false
+	}
+
+	now = time.Now().UnixMilli()
+	if callbackErr == nil {
+		// Success: mark as completed
+		_, err = s.db.ExecContext(s.ctx, `UPDATE tasks SET status = 'completed', updated_at = ? WHERE id = ?`, now, taskID)
+		if err != nil {
+			log.Error(s.ctx, "Failed to mark task as completed", "plugin", s.pluginName, "taskID", taskID, err)
+		}
+		log.Debug(s.ctx, "Task completed", "plugin", s.pluginName, "queue", queueName, "taskID", taskID)
+	} else {
+		// Failure: retry or mark as failed
+		log.Warn(s.ctx, "Task execution failed", "plugin", s.pluginName, "queue", queueName,
+			"taskID", taskID, "attempt", attempt, "maxRetries", maxRetries, "err", callbackErr)
+
+		if attempt <= maxRetries {
+			// Retry with exponential backoff: backoffMs * 2^(attempt-1)
+			backoff := qs.config.BackoffMs * int64(math.Pow(2, float64(attempt-1)))
+			nextRunAt := now + backoff
+			_, err = s.db.ExecContext(s.ctx, `
+				UPDATE tasks SET status = 'pending', next_run_at = ?, updated_at = ? WHERE id = ?
+			`, nextRunAt, now, taskID)
+			if err != nil {
+				log.Error(s.ctx, "Failed to reschedule task for retry", "plugin", s.pluginName, "taskID", taskID, err)
+			}
+
+			// Schedule a delayed signal so the worker picks up the retried task
+			// after the backoff period, rather than waiting for the next poll.
+			backoffDuration := time.Duration(backoff) * time.Millisecond
+			time.AfterFunc(backoffDuration, func() {
+				select {
+				case qs.signal <- struct{}{}:
+				default:
+				}
+			})
+		} else {
+			// Exhausted retries: mark as failed
+			_, err = s.db.ExecContext(s.ctx, `UPDATE tasks SET status = 'failed', updated_at = ? WHERE id = ?`, now, taskID)
+			if err != nil {
+				log.Error(s.ctx, "Failed to mark task as failed", "plugin", s.pluginName, "taskID", taskID, err)
+			}
+			log.Warn(s.ctx, "Task failed after all retries", "plugin", s.pluginName, "queue", queueName, "taskID", taskID)
+		}
+	}
+
+	return true
+}
+
+// revertTaskToPending puts a running task back to pending status and decrements the attempt
+// counter (used during shutdown to ensure the interrupted attempt doesn't count).
+func (s *taskQueueServiceImpl) revertTaskToPending(taskID string) {
+	now := time.Now().UnixMilli()
+	_, err := s.db.Exec(`UPDATE tasks SET status = 'pending', attempt = MAX(attempt - 1, 0), updated_at = ? WHERE id = ? AND status = 'running'`, now, taskID)
+	if err != nil {
+		log.Error("Failed to revert task to pending", "plugin", s.pluginName, "taskID", taskID, err)
+	}
+}
+
+// defaultInvokeCallback calls the plugin's nd_task_execute function.
+func (s *taskQueueServiceImpl) defaultInvokeCallback(ctx context.Context, queueName, taskID string, payload []byte, attempt int32) error {
+	s.manager.mu.RLock()
+	p, ok := s.manager.plugins[s.pluginName]
+	s.manager.mu.RUnlock()
+
+	if !ok {
+		return fmt.Errorf("plugin %s not loaded", s.pluginName)
+	}
+
+	if !hasCapability(p.capabilities, CapabilityTaskWorker) {
+		return fmt.Errorf("plugin %s lacks TaskWorker capability", s.pluginName)
+	}
+
+	input := capabilities.TaskExecuteRequest{
+		QueueName: queueName,
+		TaskID:    taskID,
+		Payload:   payload,
+		Attempt:   attempt,
+	}
+
+	result, err := callPluginFunction[capabilities.TaskExecuteRequest, capabilities.TaskExecuteResponse](ctx, p, FuncTaskWorkerCallback, input)
+	if err != nil {
+		return err
+	}
+	if result.Error != "" {
+		return fmt.Errorf("%s", result.Error)
+	}
+	return nil
+}
+
+// cleanupLoop periodically removes terminal tasks past their retention period.
+func (s *taskQueueServiceImpl) cleanupLoop() {
+	defer s.wg.Done()
+	ticker := time.NewTicker(cleanupInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-s.ctx.Done():
+			return
+		case <-ticker.C:
+			s.runCleanup()
+		}
+	}
+}
+
+// runCleanup deletes terminal tasks past their retention period.
+func (s *taskQueueServiceImpl) runCleanup() {
+	s.mu.Lock()
+	queues := make(map[string]*queueState, len(s.queues))
+	for k, v := range s.queues {
+		queues[k] = v
+	}
+	s.mu.Unlock()
+
+	now := time.Now().UnixMilli()
+	for name, qs := range queues {
+		result, err := s.db.Exec(`
+			DELETE FROM tasks WHERE queue_name = ? AND status IN ('completed', 'failed', 'cancelled') AND updated_at + ? < ?
+		`, name, qs.config.RetentionMs, now)
+		if err != nil {
+			log.Error(s.ctx, "Failed to cleanup tasks", "plugin", s.pluginName, "queue", name, err)
+			continue
+		}
+		if deleted, _ := result.RowsAffected(); deleted > 0 {
+			log.Debug(s.ctx, "Cleaned up terminal tasks", "plugin", s.pluginName, "queue", name, "deleted", deleted)
+		}
+	}
+}
+
+// Close shuts down the task queue service, stopping all workers and closing the database.
+func (s *taskQueueServiceImpl) Close() error {
+	// Cancel context to signal all goroutines
+	s.cancel()
+
+	// Wait for goroutines with timeout
+	done := make(chan struct{})
+	go func() {
+		s.wg.Wait()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(shutdownTimeout):
+		log.Warn("TaskQueue shutdown timed out", "plugin", s.pluginName)
+	}
+
+	// Mark running tasks as pending for recovery on next startup
+	if s.db != nil {
+		now := time.Now().UnixMilli()
+		_, err := s.db.Exec(`UPDATE tasks SET status = 'pending', updated_at = ? WHERE status = 'running'`, now)
+		if err != nil {
+			log.Error("Failed to reset running tasks on shutdown", "plugin", s.pluginName, err)
+		}
+
+		log.Debug("Closing plugin taskqueue", "plugin", s.pluginName)
+		return s.db.Close()
+	}
+	return nil
+}
+
+// Compile-time verification
+var _ host.TaskQueueService = (*taskQueueServiceImpl)(nil)
+var _ io.Closer = (*taskQueueServiceImpl)(nil)
--- a/plugins/host_taskqueue_test.go
+++ b/plugins/host_taskqueue_test.go
@ -0,0 +1,482 @@
+//go:build !windows
+
+package plugins
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sync/atomic"
+	"time"
+
+	"github.com/navidrome/navidrome/conf"
+	"github.com/navidrome/navidrome/conf/configtest"
+	"github.com/navidrome/navidrome/plugins/host"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("TaskQueueService", func() {
+	var tmpDir string
+	var service *taskQueueServiceImpl
+	var ctx context.Context
+	var manager *Manager
+
+	BeforeEach(func() {
+		ctx = GinkgoT().Context()
+		var err error
+		tmpDir, err = os.MkdirTemp("", "taskqueue-test-*")
+		Expect(err).ToNot(HaveOccurred())
+
+		DeferCleanup(configtest.SetupConfig())
+		conf.Server.DataFolder = tmpDir
+
+		// Create a mock manager with context
+		managerCtx, cancel := context.WithCancel(ctx)
+		manager = &Manager{
+			plugins: make(map[string]*plugin),
+			ctx:     managerCtx,
+		}
+		DeferCleanup(cancel)
+
+		service, err = newTaskQueueService("test_plugin", manager, 5)
+		Expect(err).ToNot(HaveOccurred())
+	})
+
+	AfterEach(func() {
+		if service != nil {
+			service.Close()
+		}
+		os.RemoveAll(tmpDir)
+	})
+
+	Describe("CreateQueue", func() {
+		It("creates a queue successfully", func() {
+			err := service.CreateQueue(ctx, "my-queue", host.QueueConfig{
+				Concurrency: 2,
+				MaxRetries:  3,
+				BackoffMs:   2000,
+				RetentionMs: 7200000,
+			})
+			Expect(err).ToNot(HaveOccurred())
+
+			service.mu.Lock()
+			qs, exists := service.queues["my-queue"]
+			service.mu.Unlock()
+			Expect(exists).To(BeTrue())
+			Expect(qs.config.Concurrency).To(Equal(int32(2)))
+			Expect(qs.config.MaxRetries).To(Equal(int32(3)))
+			Expect(qs.config.BackoffMs).To(Equal(int64(2000)))
+			Expect(qs.config.RetentionMs).To(Equal(int64(7200000)))
+		})
+
+		It("returns error for duplicate queue name", func() {
+			err := service.CreateQueue(ctx, "dup-queue", host.QueueConfig{})
+			Expect(err).ToNot(HaveOccurred())
+
+			err = service.CreateQueue(ctx, "dup-queue", host.QueueConfig{})
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("already exists"))
+		})
+	})
+
+	Describe("CreateQueue defaults", func() {
+		It("applies defaults for zero-value config", func() {
+			err := service.CreateQueue(ctx, "defaults-queue", host.QueueConfig{})
+			Expect(err).ToNot(HaveOccurred())
+
+			service.mu.Lock()
+			qs := service.queues["defaults-queue"]
+			service.mu.Unlock()
+			Expect(qs.config.Concurrency).To(Equal(defaultConcurrency))
+			Expect(qs.config.BackoffMs).To(Equal(defaultBackoffMs))
+			Expect(qs.config.RetentionMs).To(Equal(defaultRetentionMs))
+		})
+	})
+
+	Describe("CreateQueue clamping", func() {
+		It("clamps concurrency exceeding maxConcurrency", func() {
+			// maxConcurrency is 5; request 10
+			err := service.CreateQueue(ctx, "clamped-queue", host.QueueConfig{
+				Concurrency: 10,
+			})
+			Expect(err).ToNot(HaveOccurred())
+
+			service.mu.Lock()
+			qs := service.queues["clamped-queue"]
+			service.mu.Unlock()
+			Expect(qs.config.Concurrency).To(BeNumerically("<=", int32(5)))
+		})
+
+		It("clamps retention below minimum", func() {
+			err := service.CreateQueue(ctx, "low-retention", host.QueueConfig{
+				RetentionMs: 100, // below minRetentionMs
+			})
+			Expect(err).ToNot(HaveOccurred())
+
+			service.mu.Lock()
+			qs := service.queues["low-retention"]
+			service.mu.Unlock()
+			Expect(qs.config.RetentionMs).To(Equal(minRetentionMs))
+		})
+
+		It("clamps retention above maximum", func() {
+			err := service.CreateQueue(ctx, "high-retention", host.QueueConfig{
+				RetentionMs: 999_999_999_999, // above maxRetentionMs
+			})
+			Expect(err).ToNot(HaveOccurred())
+
+			service.mu.Lock()
+			qs := service.queues["high-retention"]
+			service.mu.Unlock()
+			Expect(qs.config.RetentionMs).To(Equal(maxRetentionMs))
+		})
+	})
+
+	Describe("Enqueue", func() {
+		BeforeEach(func() {
+			// Use a no-op callback to prevent actual execution attempts
+			service.invokeCallbackFn = func(_ context.Context, _, _ string, _ []byte, _ int32) error {
+				return nil
+			}
+			err := service.CreateQueue(ctx, "enqueue-test", host.QueueConfig{})
+			Expect(err).ToNot(HaveOccurred())
+		})
+
+		It("enqueues a task and returns task ID", func() {
+			taskID, err := service.Enqueue(ctx, "enqueue-test", []byte("payload"))
+			Expect(err).ToNot(HaveOccurred())
+			Expect(taskID).ToNot(BeEmpty())
+		})
+
+		It("returns error for non-existent queue", func() {
+			_, err := service.Enqueue(ctx, "no-such-queue", []byte("payload"))
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("does not exist"))
+		})
+	})
+
+	Describe("GetTaskStatus", func() {
+		BeforeEach(func() {
+			// Use a callback that blocks until context is cancelled so tasks stay pending
+			service.invokeCallbackFn = func(ctx context.Context, _, _ string, _ []byte, _ int32) error {
+				<-ctx.Done()
+				return ctx.Err()
+			}
+		})
+
+		It("returns pending for a new task", func() {
+			err := service.CreateQueue(ctx, "status-test", host.QueueConfig{})
+			Expect(err).ToNot(HaveOccurred())
+
+			taskID, err := service.Enqueue(ctx, "status-test", []byte("data"))
+			Expect(err).ToNot(HaveOccurred())
+
+			// The task may get picked up quickly; check initial status
+			// Since the callback blocks, it should be either pending or running
+			status, err := service.GetTaskStatus(ctx, taskID)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(status).To(BeElementOf("pending", "running"))
+		})
+
+		It("returns error for unknown task ID", func() {
+			_, err := service.GetTaskStatus(ctx, "nonexistent-id")
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("not found"))
+		})
+	})
+
+	Describe("CancelTask", func() {
+		BeforeEach(func() {
+			// Block callback so tasks stay in pending/running
+			service.invokeCallbackFn = func(ctx context.Context, _, _ string, _ []byte, _ int32) error {
+				<-ctx.Done()
+				return ctx.Err()
+			}
+		})
+
+		It("cancels a pending task", func() {
+			// Create queue with 0 concurrency via a trick: use delayMs to slow down processing
+			// Actually, just stop workers by closing and recreating without workers
+			service.Close()
+			service = nil
+
+			// Recreate without starting workers - we'll create the queue after overriding invokeCallbackFn
+			managerCtx2, cancel2 := context.WithCancel(ctx)
+			DeferCleanup(cancel2)
+			manager2 := &Manager{
+				plugins: make(map[string]*plugin),
+				ctx:     managerCtx2,
+			}
+			var err error
+			service, err = newTaskQueueService("test_plugin_cancel", manager2, 5)
+			Expect(err).ToNot(HaveOccurred())
+
+			// Block the callback so task stays pending while we try to cancel
+			service.invokeCallbackFn = func(ctx context.Context, _, _ string, _ []byte, _ int32) error {
+				time.Sleep(10 * time.Second)
+				return nil
+			}
+
+			err = service.CreateQueue(ctx, "cancel-test", host.QueueConfig{
+				Concurrency: 1,
+				DelayMs:     5000, // Large delay so worker doesn't grab it immediately
+			})
+			Expect(err).ToNot(HaveOccurred())
+
+			taskID, err := service.Enqueue(ctx, "cancel-test", []byte("cancel-me"))
+			Expect(err).ToNot(HaveOccurred())
+
+			// Cancel quickly before worker picks it up
+			err = service.CancelTask(ctx, taskID)
+			Expect(err).ToNot(HaveOccurred())
+
+			status, err := service.GetTaskStatus(ctx, taskID)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(status).To(Equal("cancelled"))
+		})
+
+		It("returns error for unknown task ID", func() {
+			err := service.CancelTask(ctx, "nonexistent-id")
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("not found"))
+		})
+
+		It("returns error for non-pending task", func() {
+			// Create a queue where tasks complete immediately
+			service.invokeCallbackFn = func(_ context.Context, _, _ string, _ []byte, _ int32) error {
+				return nil
+			}
+			err := service.CreateQueue(ctx, "completed-test", host.QueueConfig{})
+			Expect(err).ToNot(HaveOccurred())
+
+			taskID, err := service.Enqueue(ctx, "completed-test", []byte("data"))
+			Expect(err).ToNot(HaveOccurred())
+
+			// Wait for task to complete
+			Eventually(func() string {
+				status, _ := service.GetTaskStatus(ctx, taskID)
+				return status
+			}).WithTimeout(5 * time.Second).WithPolling(50 * time.Millisecond).Should(Equal("completed"))
+
+			// Try to cancel completed task
+			err = service.CancelTask(ctx, taskID)
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("cannot be cancelled"))
+		})
+	})
+
+	Describe("Worker execution", func() {
+		It("invokes callback and completes task", func() {
+			var callCount atomic.Int32
+			var receivedQueueName, receivedTaskID string
+			var receivedPayload []byte
+			var receivedAttempt int32
+
+			service.invokeCallbackFn = func(_ context.Context, queueName, taskID string, payload []byte, attempt int32) error {
+				callCount.Add(1)
+				receivedQueueName = queueName
+				receivedTaskID = taskID
+				receivedPayload = payload
+				receivedAttempt = attempt
+				return nil
+			}
+
+			err := service.CreateQueue(ctx, "worker-test", host.QueueConfig{})
+			Expect(err).ToNot(HaveOccurred())
+
+			taskID, err := service.Enqueue(ctx, "worker-test", []byte("test-payload"))
+			Expect(err).ToNot(HaveOccurred())
+
+			Eventually(func() string {
+				status, _ := service.GetTaskStatus(ctx, taskID)
+				return status
+			}).WithTimeout(5 * time.Second).WithPolling(50 * time.Millisecond).Should(Equal("completed"))
+
+			Expect(callCount.Load()).To(Equal(int32(1)))
+			Expect(receivedQueueName).To(Equal("worker-test"))
+			Expect(receivedTaskID).To(Equal(taskID))
+			Expect(receivedPayload).To(Equal([]byte("test-payload")))
+			Expect(receivedAttempt).To(Equal(int32(1)))
+		})
+	})
+
+	Describe("Retry on failure", func() {
+		It("retries and eventually fails after exhausting retries", func() {
+			var callCount atomic.Int32
+
+			service.invokeCallbackFn = func(_ context.Context, _, _ string, _ []byte, _ int32) error {
+				callCount.Add(1)
+				return fmt.Errorf("task failed")
+			}
+
+			err := service.CreateQueue(ctx, "retry-test", host.QueueConfig{
+				MaxRetries: 2,
+				BackoffMs:  10, // Very short for testing
+			})
+			Expect(err).ToNot(HaveOccurred())
+
+			taskID, err := service.Enqueue(ctx, "retry-test", []byte("retry-payload"))
+			Expect(err).ToNot(HaveOccurred())
+
+			Eventually(func() string {
+				status, _ := service.GetTaskStatus(ctx, taskID)
+				return status
+			}).WithTimeout(10 * time.Second).WithPolling(50 * time.Millisecond).Should(Equal("failed"))
+
+			// 1 initial attempt + 2 retries = 3 total calls
+			Expect(callCount.Load()).To(Equal(int32(3)))
+		})
+	})
+
+	Describe("Retry then succeed", func() {
+		It("retries and succeeds on second attempt", func() {
+			var callCount atomic.Int32
+
+			service.invokeCallbackFn = func(_ context.Context, _, _ string, _ []byte, attempt int32) error {
+				callCount.Add(1)
+				if attempt == 1 {
+					return fmt.Errorf("temporary error")
+				}
+				return nil
+			}
+
+			err := service.CreateQueue(ctx, "retry-succeed", host.QueueConfig{
+				MaxRetries: 1,
+				BackoffMs:  10, // Very short for testing
+			})
+			Expect(err).ToNot(HaveOccurred())
+
+			taskID, err := service.Enqueue(ctx, "retry-succeed", []byte("data"))
+			Expect(err).ToNot(HaveOccurred())
+
+			Eventually(func() string {
+				status, _ := service.GetTaskStatus(ctx, taskID)
+				return status
+			}).WithTimeout(10 * time.Second).WithPolling(50 * time.Millisecond).Should(Equal("completed"))
+
+			Expect(callCount.Load()).To(Equal(int32(2)))
+		})
+	})
+
+	Describe("Shutdown recovery", func() {
+		It("resets stale running tasks on CreateQueue", func() {
+			// Create a first service and queue, enqueue a task
+			service.invokeCallbackFn = func(ctx context.Context, _, _ string, _ []byte, _ int32) error {
+				<-ctx.Done()
+				return ctx.Err()
+			}
+			err := service.CreateQueue(ctx, "recovery-queue", host.QueueConfig{})
+			Expect(err).ToNot(HaveOccurred())
+
+			taskID, err := service.Enqueue(ctx, "recovery-queue", []byte("stale-task"))
+			Expect(err).ToNot(HaveOccurred())
+
+			// Wait for the task to start running
+			Eventually(func() string {
+				status, _ := service.GetTaskStatus(ctx, taskID)
+				return status
+			}).WithTimeout(5 * time.Second).WithPolling(50 * time.Millisecond).Should(Equal("running"))
+
+			// Close the service (simulates crash - tasks left in running state)
+			service.Close()
+
+			// Create a new service pointing to the same DB
+			managerCtx2, cancel2 := context.WithCancel(ctx)
+			DeferCleanup(cancel2)
+			manager2 := &Manager{
+				plugins: make(map[string]*plugin),
+				ctx:     managerCtx2,
+			}
+
+			service, err = newTaskQueueService("test_plugin", manager2, 5)
+			Expect(err).ToNot(HaveOccurred())
+
+			// Override callback to succeed
+			service.invokeCallbackFn = func(_ context.Context, _, _ string, _ []byte, _ int32) error {
+				return nil
+			}
+
+			// Re-create the queue - this should reset stale running tasks
+			// First we need to re-insert the queue row since it was from the old service
+			// Actually the queue row is already there from the first service, but
+			// CreateQueue will fail because the row exists. We need to handle this differently.
+			// The queue metadata exists in DB, but not in the new service's memory map.
+			// The schema has the queue row already. Let's delete it and re-create.
+			_, err = service.db.Exec(`DELETE FROM queues WHERE name = 'recovery-queue'`)
+			Expect(err).ToNot(HaveOccurred())
+
+			err = service.CreateQueue(ctx, "recovery-queue", host.QueueConfig{})
+			Expect(err).ToNot(HaveOccurred())
+
+			// The stale running task should now be reset to pending and eventually completed
+			Eventually(func() string {
+				status, _ := service.GetTaskStatus(ctx, taskID)
+				return status
+			}).WithTimeout(10 * time.Second).WithPolling(50 * time.Millisecond).Should(Equal("completed"))
+		})
+	})
+
+	Describe("Close", func() {
+		It("prevents subsequent operations after close", func() {
+			err := service.CreateQueue(ctx, "close-test", host.QueueConfig{})
+			Expect(err).ToNot(HaveOccurred())
+
+			service.Close()
+
+			// After close, operations should fail
+			_, err = service.Enqueue(ctx, "close-test", []byte("data"))
+			Expect(err).To(HaveOccurred())
+		})
+	})
+
+	Describe("Plugin isolation", func() {
+		It("uses separate databases for different plugins", func() {
+			managerCtx2, cancel2 := context.WithCancel(ctx)
+			DeferCleanup(cancel2)
+			manager2 := &Manager{
+				plugins: make(map[string]*plugin),
+				ctx:     managerCtx2,
+			}
+
+			service2, err := newTaskQueueService("other_plugin", manager2, 5)
+			Expect(err).ToNot(HaveOccurred())
+			defer service2.Close()
+
+			// Check that separate database files exist
+			_, err = os.Stat(filepath.Join(tmpDir, "plugins", "test_plugin", "taskqueue.db"))
+			Expect(err).ToNot(HaveOccurred())
+			_, err = os.Stat(filepath.Join(tmpDir, "plugins", "other_plugin", "taskqueue.db"))
+			Expect(err).ToNot(HaveOccurred())
+
+			// Both services should be able to create queues with the same name independently
+			service.invokeCallbackFn = func(_ context.Context, _, _ string, _ []byte, _ int32) error { return nil }
+			service2.invokeCallbackFn = func(_ context.Context, _, _ string, _ []byte, _ int32) error { return nil }
+
+			err = service.CreateQueue(ctx, "shared-name", host.QueueConfig{})
+			Expect(err).ToNot(HaveOccurred())
+			err = service2.CreateQueue(ctx, "shared-name", host.QueueConfig{})
+			Expect(err).ToNot(HaveOccurred())
+
+			// Enqueue to each and verify they work independently
+			taskID1, err := service.Enqueue(ctx, "shared-name", []byte("plugin1"))
+			Expect(err).ToNot(HaveOccurred())
+			taskID2, err := service2.Enqueue(ctx, "shared-name", []byte("plugin2"))
+			Expect(err).ToNot(HaveOccurred())
+
+			Expect(taskID1).ToNot(Equal(taskID2))
+
+			// Both should complete
+			Eventually(func() string {
+				status, _ := service.GetTaskStatus(ctx, taskID1)
+				return status
+			}).WithTimeout(5 * time.Second).WithPolling(50 * time.Millisecond).Should(Equal("completed"))
+
+			Eventually(func() string {
+				status, _ := service2.GetTaskStatus(ctx, taskID2)
+				return status
+			}).WithTimeout(5 * time.Second).WithPolling(50 * time.Millisecond).Should(Equal("completed"))
+		})
+	})
+})