From 435cfa99d25e5719ca5646005d81c3ae7a3be27b Mon Sep 17 00:00:00 2001 From: Ren0503 Date: Sat, 17 Jan 2026 10:15:50 +0700 Subject: [PATCH] feat: add methods to list and manage failed jobs in Redis - Add GetFailedJobs() to retrieve all failed jobs from Redis - Add GetFailedJob(jobId) to get a specific failed job's error - Add ClearFailedJobs() to remove all failed job records - Add scanFailedJobKeys() helper to eliminate code duplication - Reuse existing Job struct instead of creating new FailedJobInfo - Add comprehensive unit tests in failed_jobs_test.go Failed jobs are stored in Redis after exhausting all retries. These methods allow listing, inspecting, and clearing failed jobs. Closes #68 --- failed_jobs_test.go | 207 ++++++++++++++++++++++++++++++++++++++++++++ queue.go | 93 +++++++++++++++++++- 2 files changed, 299 insertions(+), 1 deletion(-) create mode 100644 failed_jobs_test.go diff --git a/failed_jobs_test.go b/failed_jobs_test.go new file mode 100644 index 0000000..dbed5b8 --- /dev/null +++ b/failed_jobs_test.go @@ -0,0 +1,207 @@ +package queue_test + +import ( + "errors" + "testing" + "time" + + "github.com/redis/go-redis/v9" + "github.com/stretchr/testify/require" + "github.com/tinh-tinh/queue/v2" +) + +func Test_GetFailedJobs(t *testing.T) { + failedQueue := queue.New("failed_jobs_test", &queue.Options{ + Connect: &redis.Options{ + Addr: "localhost:6379", + Password: "", + DB: 0, + }, + Workers: 3, + RetryFailures: 0, // No retries, so jobs fail immediately + }) + + // Clear any existing failed jobs first + err := failedQueue.ClearFailedJobs() + require.Nil(t, err) + + failedQueue.Process(func(job *queue.Job) { + job.Process(func() error { + // All jobs will fail + return errors.New("intentional failure for test") + }) + }) + + // Add multiple jobs that will fail + failedQueue.AddJob(queue.AddJobOptions{ + Id: "fail1", + Data: "value 1", + }) + failedQueue.AddJob(queue.AddJobOptions{ + Id: "fail2", + Data: "value 2", + }) + failedQueue.AddJob(queue.AddJobOptions{ + Id: "fail3", + Data: "value 3", + }) + + // Wait a bit for jobs to be processed + time.Sleep(500 * time.Millisecond) + + // Retrieve failed jobs + failedJobs, err := failedQueue.GetFailedJobs() + require.Nil(t, err) + require.Equal(t, 3, len(failedJobs)) + + // Verify job IDs are present + jobIds := make(map[string]bool) + for _, job := range failedJobs { + jobIds[job.Id] = true + require.NotEmpty(t, job.FailedReason) + require.Contains(t, job.FailedReason, "intentional failure for test") + require.Equal(t, queue.FailedStatus, job.Status) + } + require.True(t, jobIds["fail1"]) + require.True(t, jobIds["fail2"]) + require.True(t, jobIds["fail3"]) + + // Clean up + err = failedQueue.ClearFailedJobs() + require.Nil(t, err) +} + +func Test_GetFailedJob(t *testing.T) { + singleFailQueue := queue.New("single_fail_test", &queue.Options{ + Connect: &redis.Options{ + Addr: "localhost:6379", + Password: "", + DB: 0, + }, + Workers: 3, + RetryFailures: 0, + }) + + // Clear any existing failed jobs first + err := singleFailQueue.ClearFailedJobs() + require.Nil(t, err) + + singleFailQueue.Process(func(job *queue.Job) { + job.Process(func() error { + return errors.New("specific error for job " + job.Id) + }) + }) + + // Add a job that will fail + singleFailQueue.AddJob(queue.AddJobOptions{ + Id: "specific_fail", + Data: "test data", + }) + + // Wait for job to be processed + time.Sleep(500 * time.Millisecond) + + // Retrieve the specific failed job + reason, err := singleFailQueue.GetFailedJob("specific_fail") + require.Nil(t, err) + require.Contains(t, reason, "specific error for job specific_fail") + + // Try to get a non-existent failed job + _, err = singleFailQueue.GetFailedJob("non_existent") + require.NotNil(t, err) + require.Contains(t, err.Error(), "not found") + + // Clean up + err = singleFailQueue.ClearFailedJobs() + require.Nil(t, err) +} + +func Test_ClearFailedJobs(t *testing.T) { + clearQueue := queue.New("clear_test", &queue.Options{ + Connect: &redis.Options{ + Addr: "localhost:6379", + Password: "", + DB: 0, + }, + Workers: 3, + RetryFailures: 0, + }) + + // Clear any existing failed jobs first + err := clearQueue.ClearFailedJobs() + require.Nil(t, err) + + clearQueue.Process(func(job *queue.Job) { + job.Process(func() error { + return errors.New("error for clearing test") + }) + }) + + // Add multiple jobs that will fail + clearQueue.BulkAddJob([]queue.AddJobOptions{ + {Id: "clear1", Data: "value 1"}, + {Id: "clear2", Data: "value 2"}, + {Id: "clear3", Data: "value 3"}, + {Id: "clear4", Data: "value 4"}, + {Id: "clear5", Data: "value 5"}, + }) + + // Wait for jobs to be processed + time.Sleep(500 * time.Millisecond) + + // Verify failed jobs exist + failedJobs, err := clearQueue.GetFailedJobs() + require.Nil(t, err) + require.Equal(t, 5, len(failedJobs)) + + // Clear all failed jobs + err = clearQueue.ClearFailedJobs() + require.Nil(t, err) + + // Verify all failed jobs are cleared + failedJobs, err = clearQueue.GetFailedJobs() + require.Nil(t, err) + require.Equal(t, 0, len(failedJobs)) + + // Clearing again should not cause an error + err = clearQueue.ClearFailedJobs() + require.Nil(t, err) +} + +func Test_GetFailedJobs_RedisError(t *testing.T) { + // Create a queue with invalid Redis connection + invalidQueue := queue.New("redis_error_test", &queue.Options{ + Connect: &redis.Options{ + Addr: "localhost:9999", // Invalid port + Password: "", + DB: 0, + }, + Workers: 3, + RetryFailures: 0, + }) + + // Attempt to get failed jobs should return an error + // This tests that SCAN errors are propagated + _, err := invalidQueue.GetFailedJobs() + require.NotNil(t, err) + require.Contains(t, err.Error(), "failed to scan Redis keys") +} + +func Test_ClearFailedJobs_RedisError(t *testing.T) { + // Create a queue with invalid Redis connection + invalidQueue := queue.New("redis_clear_error_test", &queue.Options{ + Connect: &redis.Options{ + Addr: "localhost:9999", // Invalid port + Password: "", + DB: 0, + }, + Workers: 3, + RetryFailures: 0, + }) + + // Attempt to clear failed jobs should return an error + // This tests that SCAN errors are propagated + err := invalidQueue.ClearFailedJobs() + require.NotNil(t, err) + require.Contains(t, err.Error(), "failed to scan Redis keys") +} diff --git a/queue.go b/queue.go index ecf5775..886e939 100644 --- a/queue.go +++ b/queue.go @@ -39,7 +39,6 @@ type RateLimiter struct { Max int Duration time.Duration } - type Options struct { Connect *redis.Options Workers int @@ -476,3 +475,95 @@ func (q *Queue) log(logType LoggerType, format string, v ...any) { func (q *Queue) getKey() string { return q.cachedKey } + +// scanFailedJobKeys scans Redis for all keys matching the failed job pattern. +// Returns a slice of all matching keys or an error if the scan fails. +func (q *Queue) scanFailedJobKeys() ([]string, error) { + pattern := q.cachedKey + ":*" + var allKeys []string + + var cursor uint64 + for { + keys, nextCursor, err := q.client.Scan(q.ctx, cursor, pattern, 100).Result() + if err != nil { + return nil, fmt.Errorf("failed to scan Redis keys: %w", err) + } + + allKeys = append(allKeys, keys...) + + cursor = nextCursor + if cursor == 0 { + break + } + } + + return allKeys, nil +} + +// GetFailedJobs retrieves all failed jobs stored in Redis for this queue. +// It returns a slice of Job with Id, FailedReason, and Status populated. +// Other fields (Data, Priority, etc.) are not available as only the failure +// reason is stored in Redis. Returns an error if the Redis operation fails. +func (q *Queue) GetFailedJobs() ([]Job, error) { + keys, err := q.scanFailedJobKeys() + if err != nil { + return nil, err + } + + var failedJobs []Job + for _, key := range keys { + reason, err := q.client.Get(q.ctx, key).Result() + if err != nil { + // Only skip missing keys; propagate all other errors (network, timeout, etc.) + if err == redis.Nil { + continue + } + return nil, fmt.Errorf("failed to retrieve job data: %w", err) + } + + // Extract job ID from key (format: {prefix}{queueName}:{jobId}) + jobId := strings.TrimPrefix(key, q.cachedKey+":") + failedJobs = append(failedJobs, Job{ + Id: jobId, + FailedReason: reason, + Status: FailedStatus, + queue: q, + }) + } + + return failedJobs, nil +} + +// GetFailedJob retrieves the failure reason for a specific job by its ID. +// Returns the failure reason string or an error if the job is not found +// or if the Redis operation fails. +func (q *Queue) GetFailedJob(jobId string) (string, error) { + key := q.cachedKey + ":" + jobId + reason, err := q.client.Get(q.ctx, key).Result() + if err != nil { + if err == redis.Nil { + return "", fmt.Errorf("failed job with ID '%s' not found", jobId) + } + return "", fmt.Errorf("failed to retrieve job: %w", err) + } + return reason, nil +} + +// ClearFailedJobs removes all failed job records from Redis for this queue. +// Returns an error if the Redis operation fails. +func (q *Queue) ClearFailedJobs() error { + keysToDelete, err := q.scanFailedJobKeys() + if err != nil { + return err + } + + if len(keysToDelete) > 0 { + _, err := q.client.Del(q.ctx, keysToDelete...).Result() + if err != nil { + return fmt.Errorf("failed to delete keys: %w", err) + } + q.formatLog(LoggerInfo, "Cleared %d failed job(s)", len(keysToDelete)) + } + + return nil +}