Qualify the published four-worker queue consumer on 2.5.32

.github/queue-contention/current/composer.json

@@ -2,7 +2,7 @@
     "name": "durable-workflow/current-queue-consumer",
     "require": {
         "php": "^8.3.6",
-        "durable-workflow/workflow": "2.5.27",
+        "durable-workflow/workflow": "2.5.32",
         "laravel/framework": "13.37.0",
         "orchestra/testbench": "11.3.0",
         "phpunit/phpunit": "12.5.40"
@@ -12,6 +12,6 @@
         "sort-packages": true
     },
     "extra": {
-        "workflow-published-reference": "8ba5bec6f5fb72c6f159f4b82695f19b06017e63"
+        "workflow-published-reference": "78075b6f7e513a93db7d613bdd09b9e777fbb58a"
     }
 }

.github/queue-contention/runner.php

@@ -304,6 +304,11 @@ protected function defineEnvironment($app): void
     'rollback' => null,
     'errors' => 0,
     'repairs' => 0,
+    'job_started_ns' => null,
+    'job_pdo' => null,
+    'claim_task_id' => null,
+    'claim_started_ns' => null,
+    'claim_admissions' => 0,
 ];
 $deliveryErrorPath = $directory . '/delivery-errors.jsonl';
 $repairObservationPath = $directory . '/repair-observations.jsonl';
@@ -321,16 +326,45 @@ protected function defineEnvironment($app): void
             'stage' => 'before_update',
             'pid' => getmypid(),
             'task_id' => $task->id,
+            'task_type' => $task->getRawOriginal('task_type'),
             'status' => $task->status->value,
+            'status_before' => $task->getRawOriginal('status'),
             'repair_before' => $task->getOriginal('repair_count'),
             'repair_after' => $task->repair_count,
+            'last_dispatched_at' => $task->getRawOriginal('last_dispatched_at'),
+            'available_at' => $task->getRawOriginal('available_at'),
+            'lease_expires_at' => $task->getRawOriginal('lease_expires_at'),
+            'delivery_reference' => $reference,
             'native_row' => $row,
             'observed_unix' => time(),
         ], JSON_THROW_ON_ERROR) . "\n", FILE_APPEND | LOCK_EX);
     });
 }
 $app['events']->listen(Illuminate\Queue\Events\JobProcessing::class, static function () use ($diagnostics): void {
     $diagnostics->rollback = null;
+    $diagnostics->job_started_ns = hrtime(true);
+    $diagnostics->job_pdo = DB::connection('baseline')->getRawPdo();
+    $diagnostics->claim_task_id = null;
+    $diagnostics->claim_started_ns = null;
+    $diagnostics->claim_admissions = 0;
+});
+DB::connection('baseline')->beforeExecuting(static function (string $sql, array $bindings) use ($diagnostics): void {
+    if (! str_starts_with($sql, 'update "workflow_tasks" set "id" =') || ! isset($bindings[0])) {
+        return;
+    }
+    foreach (debug_backtrace(DEBUG_BACKTRACE_IGNORE_ARGS, 24) as $frame) {
+        if (($frame['class'] ?? null) !== Workflow\V2\Support\ActivityTaskClaimer::class
+            || ($frame['function'] ?? null) !== 'claimDetailed') {
+            continue;
+        }
+        if ($diagnostics->claim_task_id !== $bindings[0]) {
+            $diagnostics->claim_task_id = $bindings[0];
+            $diagnostics->claim_started_ns = hrtime(true);
+            $diagnostics->claim_admissions = 0;
+        }
+        $diagnostics->claim_admissions++;
+        return;
+    }
 });
 $app['events']->listen(Illuminate\Database\Events\TransactionRolledBack::class, static function ($event) use (
     $diagnostics
@@ -339,6 +373,7 @@ protected function defineEnvironment($app): void
     $diagnostics->rollback = [
         'level' => $event->connection->transactionLevel(),
         'pdo_in_transaction' => $pdo instanceof PDO ? $pdo->inTransaction() : null,
+        'same_pdo_as_job' => $pdo instanceof PDO && $pdo === $diagnostics->job_pdo,
     ];
 });
 $app['events']->listen(Illuminate\Queue\Events\JobExceptionOccurred::class, static function ($event) use (
@@ -355,6 +390,14 @@ protected function defineEnvironment($app): void
         'queue_job' => $event->job->getJobId(),
         'task_id' => transportTaskId($event->job->payload()),
         'summary' => substr($event->exception->getMessage(), 0, 384),
+        'exception_class' => $event->exception::class,
+        'native_error_info' => $event->exception instanceof PDOException ? $event->exception->errorInfo : null,
+        'job_elapsed_ms' => $diagnostics->job_started_ns === null ? null
+            : round((hrtime(true) - $diagnostics->job_started_ns) / 1_000_000, 3),
+        'claim_task_id' => $diagnostics->claim_task_id,
+        'claim_admissions' => $diagnostics->claim_admissions,
+        'claim_elapsed_ms' => $diagnostics->claim_started_ns === null ? null
+            : round((hrtime(true) - $diagnostics->claim_started_ns) / 1_000_000, 3),
         'frames' => array_slice(array_values($frames), 0, 8),
         'last_rollback' => $diagnostics->rollback,
     ], JSON_THROW_ON_ERROR) . "\n", FILE_APPEND | LOCK_EX);

.github/workflows/published-queue.yml

@@ -13,7 +13,8 @@ permissions:
 
 jobs:
   published-queue:
-    runs-on: ubuntu-latest
+    # Qualify the image's stock PHP 8.3 without a floating OS migration.
+    runs-on: ubuntu-24.04
     timeout-minutes: 8
     steps:
       - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6
@@ -46,10 +47,22 @@ jobs:
           php -r 'require ".github/queue-contention/current/vendor/autoload.php"; $manifest=json_decode(file_get_contents(".github/queue-contention/current/composer.json"), true, flags: JSON_THROW_ON_ERROR); if (Composer\InstalledVersions::getPrettyVersion("durable-workflow/workflow") !== $manifest["require"]["durable-workflow/workflow"] || Composer\InstalledVersions::getReference("durable-workflow/workflow") !== $manifest["extra"]["workflow-published-reference"]) { throw new RuntimeException("Installed package provenance differs."); }'
 
       - name: Execute the original continuous four-worker cohort
+        id: ordinary
         run: |
           sudo systemd-run --scope --quiet \
             --property=CPUQuota=200% --property=MemoryMax=2G --property=MemorySwapMax=0 \
             runuser --user "$(id -un)" -- \
             env XDEBUG_MODE=off DW_PROBE_QUEUE_WAKE=true DW_PROBE_CONTINUOUS=true \
             DW_QUALIFICATION_AUTOLOAD="$PWD/.github/queue-contention/current/vendor/autoload.php" \
             bash .github/queue-contention/current/limited-cohort.sh
+
+      - name: Capture one fresh diagnostic original after a cohort failure
+        if: ${{ failure() && steps.ordinary.outcome == 'failure' }}
+        run: |
+          sudo systemd-run --scope --quiet \
+            --property=CPUQuota=200% --property=MemoryMax=2G --property=MemorySwapMax=0 \
+            runuser --user "$(id -un)" -- \
+            env XDEBUG_MODE=off DW_PROBE_QUEUE_WAKE=true DW_PROBE_CONTINUOUS=true \
+            DW_PROBE_REPAIR_OBSERVATIONS=true \
+            DW_QUALIFICATION_AUTOLOAD="$PWD/.github/queue-contention/current/vendor/autoload.php" \
+            bash .github/queue-contention/current/limited-cohort.sh