Read recent failures with NUL-bearing PostgreSQL history

CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ## 2.12.2 - 2026-10-24
 
+- Read recent failure links and rebuild history-budget counters on PostgreSQL
+  when retained messages contain NUL bytes. Preserve canonical history and
+  exact byte counts without loading full payloads into diagnostic readers.
 - Refuse expired or settled activity callbacks under the original run lock
   before acquiring task or execution locks. Preserve committed renewals and
   the normal live-outcome path. Concurrent retention can finish without a

src/V2/Support/HistoryBudget.php

@@ -509,19 +509,23 @@ private static function isMariaDbConnection(Connection $connection): bool
      */
     private static function postgresAggregateExpressions(string $eventType, string $payload): array
     {
-        $json = "COALESCE({$payload}::jsonb::text, '[]')";
+        $metadata = PostgresJsonMetadata::textExpression($payload);
+        $json = "COALESCE({$metadata}::jsonb::text, '[]')";
+        // Each masked NUL escape occupies six canonical JSON bytes. Restore
+        // its byte difference after measuring the temporary metadata copy.
+        $nullEscapeBytes = "COALESCE(OCTET_LENGTH({$payload}::text) - OCTET_LENGTH({$metadata}), 0)";
         $jsonWithoutStrings = "regexp_replace({$json}, convert_from(decode('22285B5E225C5C5D7C5C5C2E292A22', 'hex'), 'UTF8'), '\"\"', 'g')";
         $structuralSpaces = "((CHAR_LENGTH({$jsonWithoutStrings}) - CHAR_LENGTH(REPLACE({$jsonWithoutStrings}, ':', ''))) + (CHAR_LENGTH({$jsonWithoutStrings}) - CHAR_LENGTH(REPLACE({$jsonWithoutStrings}, ',', ''))))";
         $canonicalJson = "REPLACE(REPLACE(REPLACE({$json}, E'\\\\/', '/'), "
             . "convert_from(decode('E280A8', 'hex'), 'UTF8'), convert_from(decode('5C7532303238', 'hex'), 'UTF8')), "
             . "convert_from(decode('E280A9', 'hex'), 'UTF8'), convert_from(decode('5C7532303239', 'hex'), 'UTF8'))";
-        $groupId = "{$payload}->>'parallel_group_id'";
-        $groupIdIsValid = "jsonb_typeof({$payload}::jsonb->'parallel_group_id') = 'string' AND COALESCE({$groupId}, '') <> ''";
-        $groupSize = "{$payload}->>'parallel_group_size'";
-        $groupSizeIsNumeric = "jsonb_typeof({$payload}::jsonb->'parallel_group_size') IN ('number', 'string') AND COALESCE({$groupSize}, '') ~ '^[[:space:]]*[+-]?(([0-9]+([.][0-9]*)?)|([.][0-9]+))([eE][+-]?[0-9]+)?[[:space:]]*$'";
+        $groupId = "({$metadata}::json->>'parallel_group_id')";
+        $groupIdIsValid = "jsonb_typeof({$metadata}::jsonb->'parallel_group_id') = 'string' AND COALESCE({$groupId}, '') <> ''";
+        $groupSize = "({$metadata}::json->>'parallel_group_size')";
+        $groupSizeIsNumeric = "jsonb_typeof({$metadata}::jsonb->'parallel_group_size') IN ('number', 'string') AND COALESCE({$groupSize}, '') ~ '^[[:space:]]*[+-]?(([0-9]+([.][0-9]*)?)|([.][0-9]+))([eE][+-]?[0-9]+)?[[:space:]]*$'";
 
         return [
-            "OCTET_LENGTH({$eventType}::text) + OCTET_LENGTH({$canonicalJson}) - {$structuralSpaces}",
+            "OCTET_LENGTH({$eventType}::text) + OCTET_LENGTH({$canonicalJson}) - {$structuralSpaces} + {$nullEscapeBytes}",
             $groupId,
             $groupIdIsValid,
             $groupSizeIsNumeric,

src/V2/Support/PostgresJsonMetadata.php

@@ -0,0 +1,23 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Workflow\V2\Support;
+
+final class PostgresJsonMetadata
+{
+    public static function textExpression(string $payload): string
+    {
+        // PostgreSQL reparses every string, even when extracting a different
+        // key from json. Mask actual NUL escapes only in metadata expressions.
+        // Paired backslashes remain literal, and stored history is untouched.
+        $nullEscape = <<<'REGEX'
+            (?<!\\)((?:\\\\)*)\\u0000
+            REGEX;
+        $pattern = bin2hex($nullEscape);
+        $replacement = bin2hex('\\1 ');
+
+        return "regexp_replace({$payload}::text, convert_from(decode('{$pattern}', 'hex'), 'UTF8'), "
+            . "convert_from(decode('{$replacement}', 'hex'), 'UTF8'), 'g')";
+    }
+}

src/V2/Support/RunRecentFailures.php

@@ -4,6 +4,7 @@
 
 namespace Workflow\V2\Support;
 
+use Illuminate\Database\Eloquent\Builder;
 use InvalidArgumentException;
 use Workflow\V2\Enums\HistoryEventType;
 use Workflow\V2\Models\WorkflowFailure;
@@ -66,7 +67,8 @@ public static function forRun(WorkflowRun $run, int $limit = 10): array
             ->where('workflow_run_id', $run->id);
         // Aggregate only selected identities. Successful updates and later
         // FailureHandled events must never become the primary failure link.
-        $references = (clone $query)->whereIn('payload->failure_id', $rows->modelKeys())
+        $failureId = self::failureIdExpression($query);
+        $references = (clone $query)->whereIn($connection->raw($failureId), $rows->modelKeys())
             ->whereIn('event_type', [
                 HistoryEventType::ActivityFailed->value,
                 HistoryEventType::ActivityTimedOut->value,
@@ -79,9 +81,9 @@ public static function forRun(WorkflowRun $run, int $limit = 10): array
                 HistoryEventType::WorkflowTerminated->value,
                 HistoryEventType::UpdateCompleted->value,
             ])->toBase()
-            ->select('payload->failure_id as failure_id')
+            ->selectRaw("{$failureId} AS failure_id")
             ->selectRaw('MAX(sequence) as sequence')
-            ->groupBy('payload->failure_id')
+            ->groupByRaw($failureId)
             ->get();
         $failureBySequence = $references->pluck('failure_id', 'sequence');
         $events = (clone $query)->whereIn('sequence', $failureBySequence->keys()->all())
@@ -136,4 +138,22 @@ public static function forRun(WorkflowRun $run, int $limit = 10): array
             'has_more' => $hasMore,
         ];
     }
+
+    /**
+     * @param Builder<WorkflowHistoryEvent> $query
+     */
+    private static function failureIdExpression(Builder $query): string
+    {
+        $model = $query->getModel();
+        $connection = $model->getConnection();
+        $grammar = $connection->getQueryGrammar();
+        $payload = $model->qualifyColumn('payload');
+        if ($connection->getDriverName() !== 'pgsql') {
+            return $grammar->wrap($payload . '->failure_id');
+        }
+
+        $metadata = PostgresJsonMetadata::textExpression($grammar->wrap($payload));
+
+        return "({$metadata}::json->>'failure_id')";
+    }
 }

tests/Unit/V2/HistoryBudgetTest.php

@@ -400,6 +400,43 @@ public function testBoundedFanOutUsesFirstNumericSizeForEachParallelGroup(): voi
         $this->assertFalse($run->relationLoaded('historyEvents'));
     }
 
+    #[DataProvider('largeHistorySummaryStates')]
+    public function testBoundedHistoryPreservesNullEscapesAndCanonicalByteCounts(bool $staleSummary): void
+    {
+        $run = $this->createRun();
+        $message = str_repeat("\0\0界\\u0000\\\0\"", 1_000);
+        WorkflowHistoryEvent::record($run, HistoryEventType::ActivityFailed, [
+            'failure_id' => 'failure',
+            'message' => $message,
+        ]);
+        if ($staleSummary) {
+            WorkflowRunSummary::query()->create([
+                'id' => $run->id,
+                'workflow_instance_id' => $run->workflow_instance_id,
+                'run_number' => $run->run_number,
+                'is_current_run' => true,
+                'class' => $run->workflow_class,
+                'workflow_type' => $run->workflow_type,
+                'namespace' => $run->namespace,
+                'status' => $run->status->value,
+                'status_bucket' => $run->status->statusBucket()
+->value,
+                'history_event_count' => 0,
+                'history_size_bytes' => 0,
+                'history_fan_out' => 0,
+            ]);
+        }
+        $original = DB::table('workflow_history_events')->where('workflow_run_id', $run->id)->value('payload');
+        $expected = HistoryBudget::forRun($run);
+        $run->unsetRelation('historyEvents');
+        $this->assertSame($expected, HistoryBudget::forRunBounded($run));
+        $this->assertFalse($run->relationLoaded('historyEvents'));
+        $this->assertSame(
+            $original,
+            DB::table('workflow_history_events')->where('workflow_run_id', $run->id)->value('payload')
+        );
+    }
+
     #[DataProvider('largeHistorySummaryStates')]
     public function testLargeInlineHistoryFitsDefaultMysqlSortBuffer(bool $staleSummary): void
     {

tests/Unit/V2/RunRecentFailuresTest.php

@@ -16,6 +16,35 @@
 
 final class RunRecentFailuresTest extends TestCase
 {
+    public function testSupportingEventLinksSurviveNullBytesWithoutChangingHistory(): void
+    {
+        $run = $this->createRun('null-history');
+        $identities = ['plain', 'literal\\u0000', 'slashes\\\\u0000', 'unicode-界'];
+        $message = "\0\0literal\\u0000\\\0\"{\"failure_id\":\"decoy\"}";
+        foreach ($identities as $index => $id) {
+            $this->failure($run, $id);
+            $this->event($run, $index + 1, 'ActivityFailed', [
+                'failure_id' => $id,
+                'message' => $message,
+                'details' => [
+                    'message' => $message,
+                ],
+            ]);
+        }
+        $before = DB::table('workflow_history_events')->orderBy('sequence')->pluck('payload')->all();
+        $result = RunRecentFailures::forRun($run);
+        $byIdentity = array_column($result['failures'], null, 'id');
+        foreach ($identities as $index => $id) {
+            $this->assertSame($index + 1, $byIdentity[$id]['event_sequence']);
+            $this->assertSame('ActivityFailed', $byIdentity[$id]['supporting_event']['event_type']);
+        }
+        $this->assertSame($before, DB::table('workflow_history_events')->orderBy('sequence')->pluck('payload')->all());
+        foreach (WorkflowHistoryEvent::query()->get() as $event) {
+            $this->assertSame($message, $event->payload['message']);
+            $this->assertSame($message, $event->payload['details']['message']);
+        }
+    }
+
     public function testRecentFailuresReadOnlySelectedRowsAndPrimaryEventHeaders(): void
     {
         $run = $this->createRun('selected');