From 6fc76ab6c595f717b9cebf08b51321d805c7c91d Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 24 Oct 2026 06:16:13 +0000 Subject: [PATCH 1/2] test: preserve recent failure links with NUL-bearing history --- tests/Unit/V2/RunRecentFailuresTest.php | 27 +++++++++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/tests/Unit/V2/RunRecentFailuresTest.php b/tests/Unit/V2/RunRecentFailuresTest.php index b7cdba402..98fb438da 100644 --- a/tests/Unit/V2/RunRecentFailuresTest.php +++ b/tests/Unit/V2/RunRecentFailuresTest.php @@ -16,6 +16,33 @@ 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'); From 9b16b203a1d57b2257e2e9d1c5f1feaf490f1dce Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 24 Oct 2026 06:26:05 +0000 Subject: [PATCH 2/2] fix: read PostgreSQL history metadata with NUL-bearing messages --- CHANGELOG.md | 3 ++ src/V2/Support/HistoryBudget.php | 16 +++++++---- src/V2/Support/PostgresJsonMetadata.php | 23 +++++++++++++++ src/V2/Support/RunRecentFailures.php | 26 +++++++++++++++-- tests/Unit/V2/HistoryBudgetTest.php | 37 +++++++++++++++++++++++++ tests/Unit/V2/RunRecentFailuresTest.php | 4 ++- 6 files changed, 99 insertions(+), 10 deletions(-) create mode 100644 src/V2/Support/PostgresJsonMetadata.php diff --git a/CHANGELOG.md b/CHANGELOG.md index 7d22a0c17..1f6131a23 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/V2/Support/HistoryBudget.php b/src/V2/Support/HistoryBudget.php index 8d2ec0d32..ce653f3aa 100644 --- a/src/V2/Support/HistoryBudget.php +++ b/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, diff --git a/src/V2/Support/PostgresJsonMetadata.php b/src/V2/Support/PostgresJsonMetadata.php new file mode 100644 index 000000000..e00c7a492 --- /dev/null +++ b/src/V2/Support/PostgresJsonMetadata.php @@ -0,0 +1,23 @@ +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 $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')"; + } } diff --git a/tests/Unit/V2/HistoryBudgetTest.php b/tests/Unit/V2/HistoryBudgetTest.php index 48b18ba13..92b01f0ad 100644 --- a/tests/Unit/V2/HistoryBudgetTest.php +++ b/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 { diff --git a/tests/Unit/V2/RunRecentFailuresTest.php b/tests/Unit/V2/RunRecentFailuresTest.php index 98fb438da..07974962a 100644 --- a/tests/Unit/V2/RunRecentFailuresTest.php +++ b/tests/Unit/V2/RunRecentFailuresTest.php @@ -26,7 +26,9 @@ public function testSupportingEventLinksSurviveNullBytesWithoutChangingHistory() $this->event($run, $index + 1, 'ActivityFailed', [ 'failure_id' => $id, 'message' => $message, - 'details' => ['message' => $message], + 'details' => [ + 'message' => $message, + ], ]); } $before = DB::table('workflow_history_events')->orderBy('sequence')->pluck('payload')->all();