fix(metrics): ensure throughput and duration use same time window

Fix race condition where throughput_per_minute and avg_duration_ms
were calculated from different time windows, causing mismatched metrics.

Changes:
- Add getAverageDurationInWindow() method to calculate duration from same window as throughput
- Use 60-second window for both throughput and avg_duration calculations
- Store duration/memory/CPU samples with unique "jobId:value" format to prevent overwrites
- Calculate weighted average duration across all jobs in the queue
- Separate lifetime metrics (failure_rate) from windowed metrics (throughput, avg_duration)

Technical details:
- Atomic Lua script calculates average duration from windowed samples
- Prevents race condition where jobs with identical values overwrote each other
- Ensures metric consistency by using the same Redis ZRANGEBYSCORE window

Author: sylvesterdamgaard

src/Actions/CalculateQueueMetricsAction.php

@@ -40,21 +40,34 @@ public function execute(string $connection, string $queue): void
             return;
         }
 
-        // Aggregate metrics across all job classes
+        // Aggregate metrics across all job classes for CURRENT window (last 60 seconds)
+        // This ensures throughput and avg_duration are calculated from the same time window
+        $windowSeconds = 60;
+        $throughputPerMinute = 0;
+        $totalDurationMs = 0.0;
+        $totalJobsInWindow = 0;
+
+        // Aggregate lifetime metrics for failure rate
         $totalProcessed = 0;
         $totalFailed = 0;
-        $totalDurationMs = 0.0;
         $lastProcessedAt = null;
 
         foreach ($queueJobs as $job) {
             $jobClass = $job['jobClass'];
+
+            // Get current window metrics (last 60 seconds)
+            $jobThroughput = $this->jobRepository->getThroughput($jobClass, $connection, $queue, $windowSeconds);
+            $jobAvgDuration = $this->jobRepository->getAverageDurationInWindow($jobClass, $connection, $queue, $windowSeconds);
+
+            $throughputPerMinute += $jobThroughput;
+            $totalDurationMs += ($jobAvgDuration * $jobThroughput); // Weighted by job count
+            $totalJobsInWindow += $jobThroughput;
+
+            // Get lifetime metrics for failure rate and last_processed_at
             $metrics = $this->jobRepository->getMetrics($jobClass, $connection, $queue);
 
             $totalProcessed += is_int($metrics['total_processed']) ? $metrics['total_processed'] : 0;
             $totalFailed += is_int($metrics['total_failed']) ? $metrics['total_failed'] : 0;
-            $totalDurationMs += is_float($metrics['total_duration_ms']) || is_int($metrics['total_duration_ms'])
-                ? (float) $metrics['total_duration_ms']
-                : 0.0;
 
             if ($metrics['last_processed_at'] instanceof \Carbon\Carbon) {
                 if ($lastProcessedAt === null || $metrics['last_processed_at']->greaterThan($lastProcessedAt)) {
@@ -63,24 +76,14 @@ public function execute(string $connection, string $queue): void
             }
         }
 
-        // Calculate aggregated metrics
-        $avgDuration = $totalProcessed > 0 ? $totalDurationMs / $totalProcessed : 0.0;
+        // Calculate aggregated metrics for current window
+        $avgDuration = $totalJobsInWindow > 0 ? $totalDurationMs / $totalJobsInWindow : 0.0;
+
+        // Calculate failure rate from lifetime totals
         $failureRate = ($totalProcessed + $totalFailed) > 0
             ? ($totalFailed / ($totalProcessed + $totalFailed)) * 100.0
             : 0.0;
 
-        // Calculate throughput per minute (jobs completed in last 60 seconds)
-        $throughputPerMinute = 0.0;
-        foreach ($queueJobs as $job) {
-            $jobClass = $job['jobClass'];
-            $throughputPerMinute += $this->jobRepository->getThroughput(
-                $jobClass,
-                $connection,
-                $queue,
-                60 // last 60 seconds
-            );
-        }
-
         // Store aggregated metrics
         $this->queueRepository->recordSnapshot($connection, $queue, [
             'throughput_per_minute' => $throughputPerMinute,

src/Repositories/Contracts/JobMetricsRepository.php

@@ -114,6 +114,18 @@ public function getThroughput(
         int $windowSeconds,
     ): int;
 
+    /**
+     * Get average duration for jobs completed within a specific time window.
+     *
+     * @return float Average duration in milliseconds, 0.0 if no jobs in window
+     */
+    public function getAverageDurationInWindow(
+        string $jobClass,
+        string $connection,
+        string $queue,
+        int $windowSeconds,
+    ): float;
+
     /**
      * Record when a job is queued for time-to-start tracking.
      */

src/Repositories/RedisJobMetricsRepository.php

@@ -95,27 +95,33 @@ public function recordCompletion(
             $memoryMb,
             $cpuTimeMs,
             $completedAt,
-            $ttl
+            $ttl,
+            $jobId
         ) {
             $pipe->incrementHashField($metricsKey, 'total_processed', 1);
             $pipe->incrementHashField($metricsKey, 'total_duration_ms', $durationMs);
             $pipe->incrementHashField($metricsKey, 'total_memory_mb', $memoryMb);
             $pipe->incrementHashField($metricsKey, 'total_cpu_time_ms', $cpuTimeMs);
             $pipe->setHash($metricsKey, ['last_processed_at' => $completedAt->timestamp]);
 
-            // Store duration sample (sorted set with timestamp as score)
+            // Store samples in sorted sets with timestamp as score
+            // Use a unique member format: "jobId:value" to ensure each job gets a separate entry
+            // This allows multiple jobs with the same duration/memory/cpu to be stored
+            $durationMember = $jobId . ':' . $durationMs;
             /** @var array<string, int> $durationSample */
-            $durationSample = [(string) $durationMs => (int) $completedAt->timestamp];
+            $durationSample = [$durationMember => (int) $completedAt->timestamp];
             $pipe->addToSortedSet($durationKey, $durationSample, $ttl);
 
-            // Store memory sample
+            // Store memory sample with unique member
+            $memoryMember = $jobId . ':' . $memoryMb;
             /** @var array<string, int> $memorySample */
-            $memorySample = [(string) $memoryMb => (int) $completedAt->timestamp];
+            $memorySample = [$memoryMember => (int) $completedAt->timestamp];
             $pipe->addToSortedSet($memoryKey, $memorySample, $ttl);
 
-            // Store CPU time sample
+            // Store CPU time sample with unique member
+            $cpuMember = $jobId . ':' . $cpuTimeMs;
             /** @var array<string, int> $cpuSample */
-            $cpuSample = [(string) $cpuTimeMs => (int) $completedAt->timestamp];
+            $cpuSample = [$cpuMember => (int) $completedAt->timestamp];
             $pipe->addToSortedSet($cpuKey, $cpuSample, $ttl);
 
             // Refresh TTL on metrics key
@@ -406,6 +412,56 @@ public function getThroughput(
         return $driver->eval($script, 1, $key, $windowSeconds);
     }
 
+    public function getAverageDurationInWindow(
+        string $jobClass,
+        string $connection,
+        string $queue,
+        int $windowSeconds,
+    ): float {
+        $key = $this->redis->key('durations', $connection, $queue, $jobClass);
+        $driver = $this->redis->driver();
+
+        // Use Lua script to atomically get samples within window and calculate average
+        // This ensures consistency between throughput and average duration calculations
+        $script = <<<'LUA'
+            local key = KEYS[1]
+            local windowSeconds = tonumber(ARGV[1])
+            local cutoff = redis.call('TIME')[1] - windowSeconds
+
+            -- Get all members in the window (members are "jobId:duration")
+            local samples = redis.call('ZRANGEBYSCORE', key, cutoff, '+inf')
+
+            if #samples == 0 then
+                return 0
+            end
+
+            -- Parse members to extract duration values and calculate average
+            -- Each member is formatted as "jobId:duration"
+            local sum = 0
+            local count = 0
+            for i = 1, #samples do
+                local member = samples[i]
+                local colonPos = string.find(member, ":")
+                if colonPos then
+                    local duration = string.sub(member, colonPos + 1)
+                    sum = sum + tonumber(duration)
+                    count = count + 1
+                end
+            end
+
+            if count == 0 then
+                return 0
+            end
+
+            return sum / count
+        LUA;
+
+        /** @var float */
+        $result = $driver->eval($script, 1, $key, $windowSeconds);
+
+        return (float) $result;
+    }
+
     public function recordQueuedAt(
         string $jobClass,
         string $connection,