Verified Moving all legacy quacker source code from github to cgitHEADmain

3 files changed, 920 insertions, 0 deletions

diff --git a/workerpool/workerpool.go b/workerpool/workerpool.go
new file mode 100644
index 0000000..44a3edf
--- /dev/null
+++ b/workerpool/workerpool.go
@@ -0,0 +1,262 @@
+// Package workerpool provides a generic worker pool implementation for concurrent task processing.
+package workerpool
+
+import (
+        "context"
+        "errors"
+        "fmt"
+        "maps"
+        "sync"
+        "sync/atomic"
+
+        "golang.org/x/sync/errgroup"
+)
+
+// ErrQueueFull is returned by Push when the task queue has no remaining capacity.
+var ErrQueueFull = errors.New("workerpool: task queue is full")
+
+// Task represents a unit of work that can be executed by the worker pool.
+type Task[T any] interface {
+        Execute(ctx context.Context) error
+        GetID() string
+}
+
+// WorkerPoolObserver defines callbacks for monitoring worker pool lifecycle events.
+type WorkerPoolObserver interface {
+        OnWorkerStart(workerID int)
+        OnWorkerStop(workerID int)
+        OnTaskDone(workerID int, taskID string, err error)
+}
+
+// WorkerPoolConfig holds configuration options for creating a WorkerPool instance.
+type WorkerPoolConfig struct {
+        Name          string
+        NumWorkers    int
+        TaskQueueSize int
+        Observer      WorkerPoolObserver
+        ErrorHandler  func(workerID int, taskID string, err error)
+        PanicHandler  func(workerID int, taskID string, panicValue any)
+}
+
+// Option is a functional option for configuring a WorkerPool instance.
+type Option func(*WorkerPoolConfig)
+
+// WorkerPool manages a fixed-size pool of workers that process tasks concurrently.
+type WorkerPool[T Task[T]] struct {
+        cfg       *WorkerPoolConfig
+        taskQueue chan T
+        metrics   *metrics
+        eg        *errgroup.Group
+        once      sync.Once
+        ctx       context.Context
+        started   atomic.Bool
+}
+
+type metrics struct {
+        mu             sync.RWMutex
+        TasksProcessed map[string]int
+        TasksFailed    int
+        ActiveWorkers  int
+}
+
+// MetricsSnapshot provides a point-in-time view of worker pool metrics.
+type MetricsSnapshot struct {
+        TasksProcessed map[string]int
+        TasksFailed    int
+        ActiveWorkers  int
+}
+
+// NewWorkerPool creates a new WorkerPool with the provided options.
+func NewWorkerPool[T Task[T]](ctx context.Context, opts ...Option) (*WorkerPool[T], error) {
+        cfg := &WorkerPoolConfig{
+                Name:          "workerpool",
+                NumWorkers:    1,
+                TaskQueueSize: 1,
+        }
+
+        for _, opt := range opts {
+                opt(cfg)
+        }
+
+        if cfg.NumWorkers < 1 {
+                return nil, fmt.Errorf("workerpool: number of workers must be at least 1, got %d", cfg.NumWorkers)
+        }
+
+        if cfg.TaskQueueSize < 1 {
+                return nil, fmt.Errorf("workerpool: task queue size must be at least 1, got %d", cfg.TaskQueueSize)
+        }
+
+        eg, ctx := errgroup.WithContext(ctx)
+
+        return &WorkerPool[T]{
+                cfg:       cfg,
+                taskQueue: make(chan T, cfg.TaskQueueSize),
+                ctx:       ctx,
+                eg:        eg,
+                metrics: &metrics{
+                        TasksProcessed: make(map[string]int),
+                },
+        }, nil
+}
+
+func WithName(name string) Option {
+        return func(cfg *WorkerPoolConfig) { cfg.Name = name }
+}
+
+func WithNumWorkers(n int) Option {
+        return func(cfg *WorkerPoolConfig) { cfg.NumWorkers = n }
+}
+
+func WithTaskQueueSize(size int) Option {
+        return func(cfg *WorkerPoolConfig) { cfg.TaskQueueSize = size }
+}
+
+func WithObserver(observer WorkerPoolObserver) Option {
+        return func(cfg *WorkerPoolConfig) { cfg.Observer = observer }
+}
+
+func WithErrorHandler(handler func(workerID int, taskID string, err error)) Option {
+        return func(cfg *WorkerPoolConfig) { cfg.ErrorHandler = handler }
+}
+
+func WithPanicHandler(handler func(workerID int, taskID string, panicValue any)) Option {
+        return func(cfg *WorkerPoolConfig) { cfg.PanicHandler = handler }
+}
+
+// Start launches worker goroutines. It panics if called multiple times.
+func (w *WorkerPool[T]) Start() {
+        if !w.started.CompareAndSwap(false, true) {
+                panic("workerpool: Start() called multiple times")
+        }
+
+        for i := 0; i < w.cfg.NumWorkers; i++ {
+                workerID := i
+                workerName := fmt.Sprintf("%s.worker.%d", w.cfg.Name, workerID)
+                w.eg.Go(func() error {
+                        w.metrics.mu.Lock()
+                        w.metrics.ActiveWorkers++
+                        w.metrics.mu.Unlock()
+
+                        if w.cfg.Observer != nil {
+                                safeObserve(func() { w.cfg.Observer.OnWorkerStart(workerID) })
+                        }
+
+                        defer func() {
+                                w.metrics.mu.Lock()
+                                w.metrics.ActiveWorkers--
+                                w.metrics.mu.Unlock()
+
+                                if w.cfg.Observer != nil {
+                                        safeObserve(func() { w.cfg.Observer.OnWorkerStop(workerID) })
+                                }
+                        }()
+
+                        return w.worker(workerID, workerName)
+                })
+        }
+}
+
+// Push adds a task to the queue. Returns ErrQueueFull if full, or context error if cancelled.
+func (w *WorkerPool[T]) Push(task T) error {
+        select {
+        case w.taskQueue <- task:
+                return nil
+        case <-w.ctx.Done():
+                return fmt.Errorf("workerpool %q: push: %w", w.cfg.Name, w.ctx.Err())
+        default:
+                return fmt.Errorf("workerpool %q: capacity %d: %w", w.cfg.Name, cap(w.taskQueue), ErrQueueFull)
+        }
+}
+
+// Stop closes the task queue.
+func (w *WorkerPool[T]) Stop() {
+        w.once.Do(func() {
+                close(w.taskQueue)
+        })
+}
+
+// Wait blocks until all workers finish.
+func (w *WorkerPool[T]) Wait() error {
+        return w.eg.Wait()
+}
+
+// StopAndWait closes the queue and waits for all workers to drain.
+func (w *WorkerPool[T]) StopAndWait() error {
+        w.Stop()
+        return w.Wait()
+}
+
+// GetMetrics returns a snapshot of current pool metrics.
+func (w *WorkerPool[T]) GetMetrics() MetricsSnapshot {
+        w.metrics.mu.RLock()
+        defer w.metrics.mu.RUnlock()
+
+        tasksProcessed := make(map[string]int, len(w.metrics.TasksProcessed))
+        maps.Copy(tasksProcessed, w.metrics.TasksProcessed)
+
+        return MetricsSnapshot{
+                ActiveWorkers:  w.metrics.ActiveWorkers,
+                TasksFailed:    w.metrics.TasksFailed,
+                TasksProcessed: tasksProcessed,
+        }
+}
+
+func (w *WorkerPool[T]) worker(workerID int, workerName string) error {
+        for {
+                select {
+                case task, ok := <-w.taskQueue:
+                        if !ok {
+                                return nil
+                        }
+                        w.executeSafe(workerID, workerName, task)
+                case <-w.ctx.Done():
+                        return w.ctx.Err()
+                }
+        }
+}
+
+func (w *WorkerPool[T]) executeSafe(workerID int, workerName string, task T) {
+        taskID := task.GetID()
+
+        defer func() {
+                if r := recover(); r != nil {
+                        w.recordTaskFailed()
+                        if w.cfg.PanicHandler != nil {
+                                w.cfg.PanicHandler(workerID, taskID, r)
+                        }
+                }
+        }()
+
+        if err := task.Execute(w.ctx); err != nil {
+                w.recordTaskFailed()
+                if w.cfg.Observer != nil {
+                        safeObserve(func() { w.cfg.Observer.OnTaskDone(workerID, taskID, err) })
+                }
+                if w.cfg.ErrorHandler != nil {
+                        w.cfg.ErrorHandler(workerID, taskID, err)
+                }
+                return
+        }
+
+        w.recordTaskSuccess(workerName)
+        if w.cfg.Observer != nil {
+                safeObserve(func() { w.cfg.Observer.OnTaskDone(workerID, taskID, nil) })
+        }
+}
+
+func (w *WorkerPool[T]) recordTaskSuccess(workerName string) {
+        w.metrics.mu.Lock()
+        defer w.metrics.mu.Unlock()
+        w.metrics.TasksProcessed[workerName]++
+}
+
+func (w *WorkerPool[T]) recordTaskFailed() {
+        w.metrics.mu.Lock()
+        defer w.metrics.mu.Unlock()
+        w.metrics.TasksFailed++
+}
+
+func safeObserve(fn func()) {
+        defer func() { recover() }()
+        fn()
+}
diff --git a/workerpool/workerpool_bench_test.go b/workerpool/workerpool_bench_test.go
new file mode 100644
index 0000000..1fbc672
--- /dev/null
+++ b/workerpool/workerpool_bench_test.go
@@ -0,0 +1,50 @@
+package workerpool
+
+import (
+        "context"
+        "fmt"
+        "testing"
+)
+
+type benchTask struct {
+        id string
+}
+
+func (t *benchTask) Execute(_ context.Context) error { return nil }
+func (t *benchTask) GetID() string                   { return t.id }
+
+func BenchmarkWorkerPool(b *testing.B) {
+        b.ReportAllocs()
+
+        pool, _ := NewWorkerPool[*benchTask](b.Context(),
+                WithNumWorkers(4),
+                WithTaskQueueSize(256),
+        )
+        pool.Start()
+
+        task := &benchTask{id: "bench"}
+
+        for b.Loop() {
+                pool.Push(task)
+        }
+
+        pool.StopAndWait()
+}
+
+func BenchmarkWorkerPool_Push(b *testing.B) {
+        b.ReportAllocs()
+
+        pool, _ := NewWorkerPool[*benchTask](b.Context(),
+                WithNumWorkers(8),
+                WithTaskQueueSize(1024),
+        )
+        pool.Start()
+        defer pool.StopAndWait()
+
+        b.RunParallel(func(pb *testing.PB) {
+                task := &benchTask{id: fmt.Sprintf("bench-%d", b.N)}
+                for pb.Next() {
+                        pool.Push(task)
+                }
+        })
+}
diff --git a/workerpool/workerpool_test.go b/workerpool/workerpool_test.go
new file mode 100644
index 0000000..851874b
--- /dev/null
+++ b/workerpool/workerpool_test.go
@@ -0,0 +1,608 @@
+package workerpool
+
+import (
+        "context"
+        "errors"
+        "fmt"
+        "sync"
+        "testing"
+        "time"
+)
+
+type mockTask struct {
+        id string
+        fn func(ctx context.Context) error
+}
+
+func (t *mockTask) Execute(ctx context.Context) error {
+        if t.fn != nil {
+                return t.fn(ctx)
+        }
+        return nil
+}
+
+func (t *mockTask) GetID() string {
+        return t.id
+}
+
+type taskResult struct {
+        taskID string
+        err    error
+}
+
+type mockObserver struct {
+        mu           sync.Mutex
+        workerStarts []int
+        workerStops  []int
+        tasksDone    []taskResult
+        t            *testing.T
+}
+
+func (o *mockObserver) OnWorkerStart(workerID int) {
+        o.mu.Lock()
+        defer o.mu.Unlock()
+        o.workerStarts = append(o.workerStarts, workerID)
+        if o.t != nil {
+                o.t.Logf("observer: worker %d started", workerID)
+        }
+}
+
+func (o *mockObserver) OnWorkerStop(workerID int) {
+        o.mu.Lock()
+        defer o.mu.Unlock()
+        o.workerStops = append(o.workerStops, workerID)
+        if o.t != nil {
+                o.t.Logf("observer: worker %d stopped", workerID)
+        }
+}
+
+func (o *mockObserver) OnTaskDone(workerID int, taskID string, err error) {
+        o.mu.Lock()
+        defer o.mu.Unlock()
+        o.tasksDone = append(o.tasksDone, taskResult{taskID: taskID, err: err})
+        if o.t != nil {
+                if err != nil {
+                        o.t.Logf("observer: worker %d task %s failed: %v", workerID, taskID, err)
+                } else {
+                        o.t.Logf("observer: worker %d task %s succeeded", workerID, taskID)
+                }
+        }
+}
+
+func TestNewWorkerPool(t *testing.T) {
+        t.Parallel()
+
+        tests := []struct {
+                name      string
+                opts      []Option
+                shouldErr bool
+        }{
+                {
+                        name:      "valid config with defaults",
+                        opts:      []Option{},
+                        shouldErr: false,
+                },
+                {
+                        name: "valid config with all options",
+                        opts: []Option{
+                                WithName("test-pool"),
+                                WithNumWorkers(4),
+                                WithTaskQueueSize(100),
+                        },
+                        shouldErr: false,
+                },
+                {
+                        name:      "invalid number of workers (zero)",
+                        opts:      []Option{WithNumWorkers(0)},
+                        shouldErr: true,
+                },
+                {
+                        name:      "invalid number of workers (negative)",
+                        opts:      []Option{WithNumWorkers(-1)},
+                        shouldErr: true,
+                },
+                {
+                        name:      "invalid task queue size (zero)",
+                        opts:      []Option{WithTaskQueueSize(0)},
+                        shouldErr: true,
+                },
+                {
+                        name:      "invalid task queue size (negative)",
+                        opts:      []Option{WithTaskQueueSize(-1)},
+                        shouldErr: true,
+                },
+        }
+
+        for _, test := range tests {
+                t.Run(test.name, func(t *testing.T) {
+                        t.Parallel()
+                        _, err := NewWorkerPool[*mockTask](t.Context(), test.opts...)
+                        if (err != nil) != test.shouldErr {
+                                t.Errorf("NewWorkerPool() error = %v, shouldErr %v", err, test.shouldErr)
+                        }
+                })
+        }
+}
+
+func TestWorkerPool_Observer(t *testing.T) {
+        t.Parallel()
+        ctx := t.Context()
+
+        obs := &mockObserver{t: t}
+        numWorkers := 2
+        taskCount := 5
+
+        pool, err := NewWorkerPool[*mockTask](ctx,
+                WithName("observer-test"),
+                WithNumWorkers(numWorkers),
+                WithTaskQueueSize(10),
+                WithObserver(obs),
+        )
+        if err != nil {
+                t.Fatalf("failed to create pool: %v", err)
+        }
+
+        pool.Start()
+
+        for i := range taskCount {
+                task := &mockTask{id: fmt.Sprintf("task-%d", i)}
+                if err := pool.Push(task); err != nil {
+                        t.Errorf("failed to push task: %v", err)
+                }
+        }
+
+        pool.StopAndWait()
+
+        obs.mu.Lock()
+        defer obs.mu.Unlock()
+
+        if len(obs.workerStarts) != numWorkers {
+                t.Errorf("expected %d worker starts, got %d", numWorkers, len(obs.workerStarts))
+        }
+        if len(obs.workerStops) != numWorkers {
+                t.Errorf("expected %d worker stops, got %d", numWorkers, len(obs.workerStops))
+        }
+
+        if len(obs.tasksDone) != taskCount {
+                t.Errorf("expected %d tasks done, got %d", taskCount, len(obs.tasksDone))
+        }
+        for _, r := range obs.tasksDone {
+                if r.err != nil {
+                        t.Errorf("task %s should have succeeded, got error: %v", r.taskID, r.err)
+                }
+        }
+}
+
+func TestWorkerPool_ObserverTaskFailed(t *testing.T) {
+        t.Parallel()
+        ctx := t.Context()
+
+        obs := &mockObserver{t: t}
+
+        pool, err := NewWorkerPool[*mockTask](ctx,
+                WithName("observer-fail-test"),
+                WithNumWorkers(1),
+                WithTaskQueueSize(10),
+                WithObserver(obs),
+        )
+        if err != nil {
+                t.Fatalf("failed to create pool: %v", err)
+        }
+
+        pool.Start()
+
+        task := &mockTask{
+                id: "fail-task",
+                fn: func(ctx context.Context) error {
+                        return fmt.Errorf("intentional error")
+                },
+        }
+        pool.Push(task)
+
+        pool.StopAndWait()
+
+        obs.mu.Lock()
+        defer obs.mu.Unlock()
+
+        failures := 0
+        for _, r := range obs.tasksDone {
+                if r.err != nil {
+                        failures++
+                }
+        }
+
+        if failures != 1 {
+                t.Errorf("expected 1 failed task, got %d", failures)
+        }
+        if len(obs.tasksDone) > 0 && obs.tasksDone[0].taskID != "fail-task" {
+                t.Errorf("expected failed task id 'fail-task', got %s", obs.tasksDone[0].taskID)
+        }
+}
+
+func TestWorkerPool_ObserverTaskPanic(t *testing.T) {
+        t.Parallel()
+        ctx := t.Context()
+
+        panicCaught := make(chan string, 1)
+
+        pool, err := NewWorkerPool[*mockTask](ctx,
+                WithName("observer-panic-test"),
+                WithNumWorkers(1),
+                WithTaskQueueSize(10),
+                WithPanicHandler(func(workerID int, taskID string, panicValue any) {
+                        panicCaught <- taskID
+                }),
+        )
+        if err != nil {
+                t.Fatalf("failed to create pool: %v", err)
+        }
+
+        pool.Start()
+
+        task := &mockTask{
+                id: "panic-task",
+                fn: func(ctx context.Context) error {
+                        panic("intentional panic")
+                },
+        }
+        pool.Push(task)
+
+        pool.StopAndWait()
+
+        select {
+        case taskID := <-panicCaught:
+                if taskID != "panic-task" {
+                        t.Errorf("expected panic from 'panic-task', got %s", taskID)
+                }
+        default:
+                t.Error("expected PanicHandler to be called")
+        }
+}
+
+func TestWorkerPool_ConcurrentPush(t *testing.T) {
+        t.Parallel()
+        ctx := t.Context()
+        var wg sync.WaitGroup
+        taskCount := 20
+
+        pool, err := NewWorkerPool[*mockTask](ctx,
+                WithName("concurrent-push"),
+                WithNumWorkers(3),
+                WithTaskQueueSize(taskCount),
+        )
+        if err != nil {
+                t.Fatalf("failed to create pool: %v", err)
+        }
+
+        pool.Start()
+        defer pool.StopAndWait()
+
+        for i := range taskCount {
+                wg.Add(1)
+                go func(id int) {
+                        defer wg.Done()
+                        task := &mockTask{id: fmt.Sprintf("task-%d", id)}
+                        if err := pool.Push(task); err != nil {
+                                t.Errorf("push failed: %v", err)
+                        }
+                }(i)
+        }
+
+        done := make(chan struct{})
+        go func() {
+                wg.Wait()
+                close(done)
+        }()
+
+        select {
+        case <-done:
+        case <-time.After(5 * time.Second):
+                t.Fatal("timeout waiting for concurrent push")
+        }
+}
+
+func TestWorkerPool_TaskQueueFull(t *testing.T) {
+        t.Parallel()
+        ctx := t.Context()
+
+        pool, err := NewWorkerPool[*mockTask](ctx,
+                WithName("queue-full"),
+                WithNumWorkers(1),
+                WithTaskQueueSize(2),
+        )
+        if err != nil {
+                t.Fatalf("failed to create pool: %v", err)
+        }
+
+        for i := range 2 {
+                task := &mockTask{id: fmt.Sprintf("task-%d", i)}
+                if err := pool.Push(task); err != nil {
+                        t.Errorf("unexpected error on push %d: %v", i, err)
+                }
+        }
+
+        task := &mockTask{id: "overflow-task"}
+        err = pool.Push(task)
+        if err == nil {
+                t.Fatal("expected error when queue is full, got nil")
+        }
+        if !errors.Is(err, ErrQueueFull) {
+                t.Fatalf("expected ErrQueueFull, got: %v", err)
+        }
+}
+
+func TestWorkerPool_MultipleStop(t *testing.T) {
+        t.Parallel()
+        ctx := t.Context()
+
+        pool, err := NewWorkerPool[*mockTask](ctx,
+                WithName("multiple-stop"),
+                WithNumWorkers(2),
+                WithTaskQueueSize(10),
+        )
+        if err != nil {
+                t.Fatalf("failed to create pool: %v", err)
+        }
+
+        pool.Start()
+        pool.Stop()
+
+        func() {
+                defer func() {
+                        if r := recover(); r != nil {
+                                t.Errorf("Stop() panicked on second call: %v", r)
+                        }
+                }()
+                pool.Stop()
+        }()
+
+        pool.Wait()
+}
+
+func TestWorkerPool_MetricsThreadSafe(t *testing.T) {
+        t.Parallel()
+        ctx := t.Context()
+        var wg sync.WaitGroup
+        taskCount := 20
+
+        pool, err := NewWorkerPool[*mockTask](ctx,
+                WithName("metrics-safe"),
+                WithNumWorkers(10),
+                WithTaskQueueSize(taskCount),
+        )
+        if err != nil {
+                t.Fatalf("failed to create pool: %v", err)
+        }
+
+        pool.Start()
+        defer pool.StopAndWait()
+
+        for i := range taskCount {
+                wg.Add(1)
+                go func(id int) {
+                        defer wg.Done()
+                        task := &mockTask{
+                                id: fmt.Sprintf("task-%d", id),
+                                fn: func(ctx context.Context) error {
+                                        time.Sleep(time.Millisecond)
+                                        return nil
+                                },
+                        }
+                        pool.Push(task)
+                }(i)
+        }
+
+        concurrentReads := 100
+        for range concurrentReads {
+                wg.Add(1)
+                go func() {
+                        defer wg.Done()
+                        for range 10 {
+                                pool.GetMetrics()
+                        }
+                }()
+        }
+
+        wg.Wait()
+}
+
+func TestWorkerPool_ContextCancellation(t *testing.T) {
+        t.Parallel()
+        ctx, cancel := context.WithCancel(t.Context())
+
+        pool, err := NewWorkerPool[*mockTask](ctx,
+                WithName("context-cancel"),
+                WithNumWorkers(2),
+                WithTaskQueueSize(10),
+        )
+        if err != nil {
+                t.Fatalf("failed to create pool: %v", err)
+        }
+
+        pool.Start()
+
+        for i := range 5 {
+                task := &mockTask{id: fmt.Sprintf("task-%d", i)}
+                pool.Push(task)
+        }
+
+        cancel()
+
+        err = pool.Wait()
+        if err != nil && !errors.Is(err, context.Canceled) {
+                t.Errorf("expected context.Canceled, got: %v", err)
+        }
+}
+
+func TestWorkerPool_ErrorHandler(t *testing.T) {
+        t.Parallel()
+        ctx := t.Context()
+
+        errorHandlerCalled := make(chan struct{})
+
+        pool, err := NewWorkerPool[*mockTask](ctx,
+                WithName("error-handler-test"),
+                WithNumWorkers(1),
+                WithTaskQueueSize(10),
+                WithErrorHandler(func(workerID int, taskID string, err error) {
+                        t.Logf("ErrorHandler called: worker=%d, task=%s, err=%v", workerID, taskID, err)
+                        close(errorHandlerCalled)
+                }),
+        )
+        if err != nil {
+                t.Fatalf("failed to create pool: %v", err)
+        }
+
+        pool.Start()
+
+        task := &mockTask{
+                id: "error-task",
+                fn: func(ctx context.Context) error {
+                        return fmt.Errorf("task error")
+                },
+        }
+        pool.Push(task)
+
+        select {
+        case <-errorHandlerCalled:
+        case <-time.After(time.Second):
+                t.Fatal("ErrorHandler was not called within timeout")
+        }
+
+        pool.StopAndWait()
+}
+
+func TestWorkerPool_PanicHandler(t *testing.T) {
+        t.Parallel()
+        ctx := t.Context()
+
+        panicHandlerCalled := make(chan struct{})
+
+        pool, err := NewWorkerPool[*mockTask](ctx,
+                WithName("panic-handler-test"),
+                WithNumWorkers(1),
+                WithTaskQueueSize(10),
+                WithPanicHandler(func(workerID int, taskID string, panicValue any) {
+                        t.Logf("PanicHandler called: worker=%d, task=%s, panic=%v", workerID, taskID, panicValue)
+                        close(panicHandlerCalled)
+                }),
+        )
+        if err != nil {
+                t.Fatalf("failed to create pool: %v", err)
+        }
+
+        pool.Start()
+
+        task := &mockTask{
+                id: "panic-task",
+                fn: func(ctx context.Context) error {
+                        panic("task panic")
+                },
+        }
+        pool.Push(task)
+
+        select {
+        case <-panicHandlerCalled:
+        case <-time.After(time.Second):
+                t.Fatal("PanicHandler was not called within timeout")
+        }
+
+        pool.StopAndWait()
+}
+
+func TestWorkerPool_PushContextCancelled(t *testing.T) {
+        t.Parallel()
+        ctx, cancel := context.WithCancel(t.Context())
+
+        pool, err := NewWorkerPool[*mockTask](ctx,
+                WithName("push-ctx-cancelled"),
+                WithNumWorkers(1),
+                WithTaskQueueSize(1),
+        )
+        if err != nil {
+                t.Fatalf("failed to create pool: %v", err)
+        }
+
+        task1 := &mockTask{id: "task-1"}
+        if err := pool.Push(task1); err != nil {
+                t.Fatalf("failed to fill queue: %v", err)
+        }
+
+        cancel()
+
+        task2 := &mockTask{id: "task-2"}
+        err = pool.Push(task2)
+
+        if err == nil {
+                t.Fatal("expected error when pushing to full queue with cancelled context")
+        }
+
+        if !errors.Is(err, ErrQueueFull) && !errors.Is(err, context.Canceled) {
+                t.Fatalf("expected ErrQueueFull or context.Canceled, got: %v", err)
+        }
+}
+
+func TestWorkerPool_NoObserver(t *testing.T) {
+        t.Parallel()
+        ctx := t.Context()
+
+        pool, err := NewWorkerPool[*mockTask](ctx,
+                WithName("no-observer"),
+                WithNumWorkers(2),
+                WithTaskQueueSize(10),
+        )
+        if err != nil {
+                t.Fatalf("failed to create pool: %v", err)
+        }
+
+        pool.Start()
+
+        task1 := &mockTask{id: "success-task"}
+        pool.Push(task1)
+
+        task2 := &mockTask{
+                id: "fail-task",
+                fn: func(ctx context.Context) error {
+                        return fmt.Errorf("error")
+                },
+        }
+        pool.Push(task2)
+
+        task3 := &mockTask{
+                id: "panic-task",
+                fn: func(ctx context.Context) error {
+                        panic("panic")
+                },
+        }
+        pool.Push(task3)
+
+        pool.StopAndWait()
+
+        metrics := pool.GetMetrics()
+        if metrics.TasksFailed != 2 {
+                t.Errorf("expected 2 failed tasks, got %d", metrics.TasksFailed)
+        }
+}
+
+func TestWorkerPool_StartMultipleTimes(t *testing.T) {
+        t.Parallel()
+
+        pool, err := NewWorkerPool[*mockTask](t.Context(),
+                WithNumWorkers(1),
+                WithTaskQueueSize(1),
+        )
+        if err != nil {
+                t.Fatalf("failed to create pool: %v", err)
+        }
+
+        pool.Start()
+        defer pool.StopAndWait()
+
+        defer func() {
+                if r := recover(); r == nil {
+                        t.Error("expected panic when calling Start() multiple times")
+                }
+        }()
+
+        pool.Start()
+}