feat: add errors.Join for batch operations and benchmark tests

- Enhance batchOperation to return aggregated errors using errors.Join
- Add benchmark tests using Go 1.24+ testing.B.Loop() pattern
- Update dependencies to latest versions

Author: kaptinlin

.github/workflows/ci.yml

@@ -23,7 +23,7 @@ jobs:
           - 6379:6379
 
     steps:
-    - uses: actions/checkout@v5
+    - uses: actions/checkout@v6
 
     - name: Set up Go
       uses: actions/setup-go@v6
@@ -37,7 +37,7 @@ jobs:
   lint:
     runs-on: ubuntu-latest
     steps:
-    - uses: actions/checkout@v5
+    - uses: actions/checkout@v6
 
     - name: Set up Go
       uses: actions/setup-go@v6

.golangci.version

@@ -1 +1 @@
-2.4.0
+2.7.2

go.mod

@@ -317,27 +317,32 @@ func (s *Manager) RunJobsByState(queue string, state JobState) (int, error) {
 	return count, nil
 }
 
-// batchOperation performs a batch operation on items and returns succeeded and failed items.
-func batchOperation[T any](items []T, operation func(T) error) (succeeded, failed []T) {
+// batchOperation performs a batch operation on items and returns succeeded, failed items, and aggregated errors.
+func batchOperation[T any](items []T, operation func(T) error) (succeeded, failed []T, err error) {
 	succeeded = make([]T, 0, len(items))
 	failed = make([]T, 0, len(items))
+	var errs []error
 
 	for _, item := range items {
-		if err := operation(item); err != nil {
+		if opErr := operation(item); opErr != nil {
 			failed = append(failed, item)
+			errs = append(errs, opErr)
 		} else {
 			succeeded = append(succeeded, item)
 		}
 	}
-	return succeeded, failed
+
+	if len(errs) > 0 {
+		return succeeded, failed, errors.Join(errs...)
+	}
+	return succeeded, failed, nil
 }
 
 // BatchRunJobs triggers immediate execution of multiple jobs identified by their IDs.
 func (s *Manager) BatchRunJobs(queue string, jobIDs []string) ([]string, []string, error) {
-	succeeded, failed := batchOperation(jobIDs, func(jobID string) error {
+	return batchOperation(jobIDs, func(jobID string) error {
 		return s.Inspector.RunTask(queue, jobID)
 	})
-	return succeeded, failed, nil
 }
 
 // ArchiveJob moves a job with the specified ID to the archive.
@@ -394,10 +399,9 @@ func (s *Manager) ArchiveJobsByState(queue string, state JobState) (int, error)
 
 // BatchArchiveJobs archives multiple jobs identified by their IDs.
 func (s *Manager) BatchArchiveJobs(queue string, jobIDs []string) ([]string, []string, error) {
-	succeeded, failed := batchOperation(jobIDs, func(jobID string) error {
+	return batchOperation(jobIDs, func(jobID string) error {
 		return s.Inspector.ArchiveTask(queue, jobID)
 	})
-	return succeeded, failed, nil
 }
 
 // CancelJob cancels a job with the specified ID.
@@ -447,10 +451,9 @@ func (s *Manager) CancelActiveJobs(queue string, size, page int) (int, error) {
 
 // BatchCancelJobs cancels multiple jobs identified by their IDs.
 func (s *Manager) BatchCancelJobs(jobIDs []string) ([]string, []string, error) {
-	succeeded, failed := batchOperation(jobIDs, func(jobID string) error {
+	return batchOperation(jobIDs, func(jobID string) error {
 		return s.Inspector.CancelProcessing(jobID)
 	})
-	return succeeded, failed, nil
 }
 
 // DeleteJob deletes a job with the specified ID from its queue.
@@ -501,10 +504,9 @@ func (s *Manager) DeleteJobsByState(queue string, state JobState) (int, error) {
 
 // BatchDeleteJobs deletes multiple jobs identified by their IDs.
 func (s *Manager) BatchDeleteJobs(queue string, jobIDs []string) ([]string, []string, error) {
-	succeeded, failed := batchOperation(jobIDs, func(jobID string) error {
+	return batchOperation(jobIDs, func(jobID string) error {
 		return s.Inspector.DeleteTask(queue, jobID)
 	})
-	return succeeded, failed, nil
 }
 
 // RunAggregatingJobs triggers all aggregating jobs to run immediately in a specified queue and group.

go.sum

@@ -0,0 +1,118 @@
+package tests
+
+import (
+	"context"
+	"testing"
+
+	"github.com/kaptinlin/queue"
+)
+
+func BenchmarkJobCreation(b *testing.B) {
+	payload := map[string]interface{}{"key": "value", "count": 123}
+
+	for b.Loop() {
+		_ = queue.NewJob("test_job", payload)
+	}
+}
+
+func BenchmarkJobCreationWithOptions(b *testing.B) {
+	payload := map[string]interface{}{"key": "value", "count": 123}
+
+	for b.Loop() {
+		_ = queue.NewJob("test_job", payload,
+			queue.WithQueue("critical"),
+			queue.WithMaxRetries(3),
+		)
+	}
+}
+
+func BenchmarkJobConvertToAsynqTask(b *testing.B) {
+	payload := map[string]interface{}{"key": "value", "count": 123}
+	job := queue.NewJob("test_job", payload, queue.WithQueue("default"))
+
+	for b.Loop() {
+		_, _, err := job.ConvertToAsynqTask()
+		if err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
+func BenchmarkHandlerCreation(b *testing.B) {
+	handlerFunc := func(_ context.Context, _ *queue.Job) error {
+		return nil
+	}
+
+	for b.Loop() {
+		_ = queue.NewHandler("test_job", handlerFunc)
+	}
+}
+
+func BenchmarkHandlerProcess(b *testing.B) {
+	handler := queue.NewHandler("test_job", func(_ context.Context, _ *queue.Job) error {
+		return nil
+	})
+
+	job := queue.NewJob("test_job", map[string]interface{}{"key": "value"})
+	ctx := context.Background()
+
+	for b.Loop() {
+		err := handler.Process(ctx, job)
+		if err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
+func BenchmarkJobDecodePayload(b *testing.B) {
+	type TestPayload struct {
+		Key   string `json:"key"`
+		Count int    `json:"count"`
+	}
+
+	payload := map[string]interface{}{"key": "value", "count": 123}
+	job := queue.NewJob("test_job", payload)
+
+	for b.Loop() {
+		var decoded TestPayload
+		err := job.DecodePayload(&decoded)
+		if err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
+func BenchmarkClientEnqueue(b *testing.B) {
+	redisConfig := getRedisConfig()
+	client, err := queue.NewClient(redisConfig)
+	if err != nil {
+		b.Skipf("Redis not available: %v", err)
+	}
+	defer func() { _ = client.Stop() }()
+
+	payload := map[string]interface{}{"key": "value"}
+
+	for b.Loop() {
+		_, err := client.Enqueue("benchmark_job", payload)
+		if err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
+func BenchmarkClientEnqueueJob(b *testing.B) {
+	redisConfig := getRedisConfig()
+	client, err := queue.NewClient(redisConfig)
+	if err != nil {
+		b.Skipf("Redis not available: %v", err)
+	}
+	defer func() { _ = client.Stop() }()
+
+	for b.Loop() {
+		job := queue.NewJob("benchmark_job", map[string]interface{}{"key": "value"})
+		_, err := client.EnqueueJob(job)
+		if err != nil {
+			b.Fatal(err)
+		}
+	}
+}