fix(redis): Redis set operations require spread operator for variadic args

Fixed critical bug in PipelineWrapper where addToSet() and removeFromSet()
were passing arrays directly to PhpRedis sadd()/srem() commands instead of
using spread operator. This caused job discovery to fail completely.

Changes:
- PipelineWrapper::addToSet(): Changed `sadd($key, $members)` to `sadd($key, ...$members)`
- PipelineWrapper::removeFromSet(): Changed `srem($key, $members)` to `srem($key, ...$members)`
- CalculateQueueMetricsTest: Added recordStart() calls before recordCompletion()
- CalculateQueueMetricsTest: Added Redis flush in beforeEach() for clean state
- CalculateQueueMetricsTest: Changed float assertions from toBe() to toEqualWithDelta()

Root Cause:
PhpRedis expects variadic arguments for set operations:
- sadd(key, member1, member2, ...) ✅
- sadd(key, [member1, member2]) ❌

Impact:
This bug prevented all job discovery tracking, causing listJobs() to return
empty arrays even though job metrics were being recorded correctly. This
made queue-level metric aggregation fail with 0 values.

Author: sylvesterdamgaard

src/Support/PipelineWrapper.php

@@ -101,7 +101,7 @@ public function removeFromSortedSet(string $key, string $member): void
     public function addToSet(string $key, array $members): void
     {
         if (! empty($members)) {
-            $this->pipe->sadd($key, $members);
+            $this->pipe->sadd($key, ...$members);
         }
     }
 
@@ -116,7 +116,7 @@ public function getSetMembers(string $key): mixed
     public function removeFromSet(string $key, array $members): void
     {
         if (! empty($members)) {
-            $this->pipe->srem($key, $members);
+            $this->pipe->srem($key, ...$members);
         }
     }
 

tests/Feature/CalculateQueueMetricsTest.php

@@ -16,6 +16,9 @@
     config()->set('queue-metrics.enabled', true);
     config()->set('queue-metrics.storage.driver', 'redis');
     config()->set('queue-metrics.storage.connection', 'default');
+
+    // Flush Redis before each test to ensure clean state
+    \Illuminate\Support\Facades\Redis::connection('default')->flushdb();
 });
 
 it('calculates queue metrics from job metrics', function () {
@@ -28,34 +31,39 @@
     $jobRepo = app(JobMetricsRepository::class);
     $queueRepo = app(QueueMetricsRepository::class);
 
-    // Mark jobs as discovered
-    $jobRepo->markJobDiscovered($jobClass1, $connection, $queue);
-    $jobRepo->markJobDiscovered($jobClass2, $connection, $queue);
-
     // Mark queue as discovered
     $queueRepo->markQueueDiscovered($connection, $queue);
 
     // Record job completions for job 1: 10 jobs, 100ms avg duration
+    // Note: recordStart() now handles job discovery atomically
     for ($i = 0; $i < 10; $i++) {
+        $jobId = "job-1-{$i}";
+        $jobRepo->recordStart($jobId, $jobClass1, $connection, $queue, \Carbon\Carbon::now());
         $jobRepo->recordCompletion(
+            jobId: $jobId,
             jobClass: $jobClass1,
             connection: $connection,
             queue: $queue,
             durationMs: 100.0,
             memoryMb: 10.0,
             cpuTimeMs: 50.0,
+            completedAt: \Carbon\Carbon::now(),
         );
     }
 
     // Record job completions for job 2: 5 jobs, 200ms avg duration
     for ($i = 0; $i < 5; $i++) {
+        $jobId = "job-2-{$i}";
+        $jobRepo->recordStart($jobId, $jobClass2, $connection, $queue, \Carbon\Carbon::now());
         $jobRepo->recordCompletion(
+            jobId: $jobId,
             jobClass: $jobClass2,
             connection: $connection,
             queue: $queue,
             durationMs: 200.0,
             memoryMb: 15.0,
             cpuTimeMs: 75.0,
+            completedAt: \Carbon\Carbon::now(),
         );
     }
 
@@ -70,7 +78,7 @@
 
     // Weighted average duration: (10 * 100 + 5 * 200) / 15 = 133.33ms
     $expectedAvgDuration = (10 * 100.0 + 5 * 200.0) / 15;
-    expect($metrics['avg_duration'])->toBe($expectedAvgDuration);
+    expect($metrics['avg_duration'])->toEqualWithDelta($expectedAvgDuration, 0.01);
 
     // Throughput should be sum of both job classes
     expect($metrics['throughput_per_minute'])->toBeGreaterThan(0.0);
@@ -107,29 +115,34 @@
     $jobRepo = app(JobMetricsRepository::class);
     $queueRepo = app(QueueMetricsRepository::class);
 
-    $jobRepo->markJobDiscovered($jobClass, $connection, $queue);
     $queueRepo->markQueueDiscovered($connection, $queue);
 
     // Record 7 successful completions
+    // Note: recordStart() now handles job discovery atomically
     for ($i = 0; $i < 7; $i++) {
+        $jobId = "job-success-{$i}";
+        $jobRepo->recordStart($jobId, $jobClass, $connection, $queue, \Carbon\Carbon::now());
         $jobRepo->recordCompletion(
+            jobId: $jobId,
             jobClass: $jobClass,
             connection: $connection,
             queue: $queue,
             durationMs: 100.0,
             memoryMb: 10.0,
             cpuTimeMs: 50.0,
+            completedAt: \Carbon\Carbon::now(),
         );
     }
 
     // Record 3 failures
     for ($i = 0; $i < 3; $i++) {
         $jobRepo->recordFailure(
+            jobId: "job-failure-{$i}",
             jobClass: $jobClass,
             connection: $connection,
             queue: $queue,
             exception: 'Test exception',
-            failedAt: time(),
+            failedAt: \Carbon\Carbon::now(),
         );
     }
 
@@ -156,16 +169,20 @@
     foreach ($queues as $q) {
         $queueRepo->markQueueDiscovered($q['connection'], $q['queue']);
         $jobClass = 'App\\Jobs\\TestJob'.ucfirst($q['queue']);
-        $jobRepo->markJobDiscovered($jobClass, $q['connection'], $q['queue']);
 
         // Record some completions
+        // Note: recordStart() now handles job discovery atomically
+        $jobId = "job-{$q['queue']}-1";
+        $jobRepo->recordStart($jobId, $jobClass, $q['connection'], $q['queue'], \Carbon\Carbon::now());
         $jobRepo->recordCompletion(
+            jobId: $jobId,
             jobClass: $jobClass,
             connection: $q['connection'],
             queue: $q['queue'],
             durationMs: 150.0,
             memoryMb: 12.0,
             cpuTimeMs: 60.0,
+            completedAt: \Carbon\Carbon::now(),
         );
     }
 
@@ -178,7 +195,7 @@
     foreach ($queues as $q) {
         $metrics = $queueRepo->getLatestMetrics($q['connection'], $q['queue']);
         expect($metrics)->not()->toBeEmpty();
-        expect($metrics['avg_duration'])->toBe(150.0);
+        expect($metrics['avg_duration'])->toEqualWithDelta(150.0, 0.01);
     }
 })->group('redis');
 
@@ -191,15 +208,20 @@
     $queueRepo = app(QueueMetricsRepository::class);
 
     $queueRepo->markQueueDiscovered($connection, $queue);
-    $jobRepo->markJobDiscovered($jobClass, $connection, $queue);
 
+    // Record job completion
+    // Note: recordStart() now handles job discovery atomically
+    $jobId = 'job-specific-1';
+    $jobRepo->recordStart($jobId, $jobClass, $connection, $queue, \Carbon\Carbon::now());
     $jobRepo->recordCompletion(
+        jobId: $jobId,
         jobClass: $jobClass,
         connection: $connection,
         queue: $queue,
         durationMs: 250.0,
         memoryMb: 20.0,
         cpuTimeMs: 100.0,
+        completedAt: \Carbon\Carbon::now(),
     );
 
     // Execute command for specific queue
@@ -211,7 +233,7 @@
     expect($exitCode)->toBe(0);
 
     $metrics = $queueRepo->getLatestMetrics($connection, $queue);
-    expect($metrics['avg_duration'])->toBe(250.0);
+    expect($metrics['avg_duration'])->toEqualWithDelta(250.0, 0.01);
 })->group('redis');
 
 it('command fails when queue specified without connection', function () {