queue_test.go

package goqueue

import (
	"context"
	"errors"
	"sync/atomic"
	"testing"
	"time"
)

func silenceLogs() { Log.Info = func(string, ...any) {}; Log.Error = func(string, ...any) {} }

func TestQueueRunsAllJobs(t *testing.T) {
	silenceLogs()
	q := New("t", 4)
	q.Start(2)

	var count int32
	for range 10 {
		if err := q.Add(Job{Name: "j", Action: func(context.Context) error {
			atomic.AddInt32(&count, 1)
			return nil
		}}); err != nil {
			t.Fatalf("Add: %v", err)
		}
	}
	q.Close()
	q.Wait()

	if got := atomic.LoadInt32(&count); got != 10 {
		t.Fatalf("ran %d jobs, want 10", got)
	}
}

func TestAddAfterStopReturnsError(t *testing.T) {
	silenceLogs()
	q := New("t", 1)
	q.Start(1)
	q.Stop()
	q.Wait()

	err := q.Add(Job{Name: "j", Action: func(context.Context) error { return nil }})
	if !errors.Is(err, ErrQueueStopped) {
		t.Fatalf("got %v, want ErrQueueStopped", err)
	}
}

func TestStopCancelsJobContext(t *testing.T) {
	silenceLogs()
	q := New("t", 1)
	q.Start(1)

	started := make(chan struct{})
	done := make(chan error, 1)
	_ = q.Add(Job{Name: "slow", Action: func(ctx context.Context) error {
		close(started)
		<-ctx.Done()
		done <- ctx.Err()
		return ctx.Err()
	}})

	<-started
	q.Stop()

	select {
	case err := <-done:
		if !errors.Is(err, context.Canceled) {
			t.Fatalf("job ctx err %v, want Canceled", err)
		}
	case <-time.After(time.Second):
		t.Fatal("job did not observe cancellation")
	}
	q.Wait()
}

func TestDoubleCloseSafe(t *testing.T) {
	silenceLogs()
	q := New("t", 1)
	q.Start(1)
	q.Close()
	q.Close()
	q.Wait()
}

func TestJobErrorDoesNotStopWorker(t *testing.T) {
	silenceLogs()
	q := New("t", 2)
	q.Start(1)

	var ok int32
	_ = q.Add(Job{Name: "bad", Action: func(context.Context) error { return errors.New("boom") }})
	_ = q.Add(Job{Name: "good", Action: func(context.Context) error {
		atomic.StoreInt32(&ok, 1)
		return nil
	}})
	q.Close()
	q.Wait()

	if atomic.LoadInt32(&ok) != 1 {
		t.Fatal("second job did not run after first failed")
	}
}