From c0635adec8db87498365a26ff64b1b83bf9e4238 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Wed, 21 Oct 2026 03:18:49 +0000 Subject: [PATCH 1/2] Classify recovered SQLite queue dispatch contention in tracing tests --- .../Support/TracingQueueErrorDiagnostics.php | 30 ++++++++++ .../V2/LargeFanOutExecutionTracingTest.php | 3 +- .../V2/TracingQueueErrorDiagnosticsTest.php | 56 +++++++++++++++++++ 3 files changed, 88 insertions(+), 1 deletion(-) diff --git a/tests/Support/TracingQueueErrorDiagnostics.php b/tests/Support/TracingQueueErrorDiagnostics.php index 8260b222..0d68c298 100644 --- a/tests/Support/TracingQueueErrorDiagnostics.php +++ b/tests/Support/TracingQueueErrorDiagnostics.php @@ -52,4 +52,34 @@ public static function isFailedJobLogContention(array $error): bool && ($frame['file'] ?? null) === 'WorkCommand.php' && ($frame['function'] ?? null) === 'logFailedJob') !== []; } + + public static function isQueueDispatchContention(array $error): bool + { + $code = $error['native_code'] ?? null; + if ( + ($error['error_class'] ?? null) !== QueryException::class + || ! is_int($code) || $code < 0 || ! in_array($code & 255, [5, 6], true) + || ($error['operation'] ?? null) !== 'insert' + || ($error['transaction_level'] ?? null) !== 0 + || ($error['pdo_in_transaction'] ?? null) !== false + || ! array_key_exists('ack_fence', $error) || $error['ack_fence'] !== null + || ($error['statement_truncated'] ?? null) !== false + || ($error['statement'] ?? null) !== 'insert into "jobs" ' + . '("queue", "attempts", "reserved_at", "available_at", "created_at", "payload") values (?, ?, ?, ?, ?, ?)' + ) { + return false; + } + + $frames = $error['stage_frames'] ?? []; + if (! is_array($frames)) { + return false; + } + + return array_filter($frames, static fn (mixed $frame): bool => is_array($frame) + && ($frame['file'] ?? null) === 'DatabaseQueue.php' + && ($frame['function'] ?? null) === 'insertGetId') !== [] + && array_filter($frames, static fn (mixed $frame): bool => is_array($frame) + && ($frame['file'] ?? null) === 'DatabaseQueue.php' + && ($frame['function'] ?? null) === 'pushToDatabase') !== []; + } } diff --git a/tests/Unit/V2/LargeFanOutExecutionTracingTest.php b/tests/Unit/V2/LargeFanOutExecutionTracingTest.php index 23bec5c2..7c533c2f 100644 --- a/tests/Unit/V2/LargeFanOutExecutionTracingTest.php +++ b/tests/Unit/V2/LargeFanOutExecutionTracingTest.php @@ -181,7 +181,8 @@ private static function assertQueuePressureError(array $error, bool $reported, s self::assertIsInt($error['native_code'], $diagnostic); self::assertContains($error['native_code'] & 255, [5, 6], $diagnostic); self::assertTrue(in_array($error['operation'], ['update', 'delete'], true) - || ($reported && TracingQueueErrorDiagnostics::isFailedJobLogContention($error)), $diagnostic); + || ($reported && (TracingQueueErrorDiagnostics::isFailedJobLogContention($error) + || TracingQueueErrorDiagnostics::isQueueDispatchContention($error))), $diagnostic); self::assertSame(0, $error['transaction_level'], $diagnostic); self::assertFalse($error['pdo_in_transaction'], $diagnostic); } diff --git a/tests/Unit/V2/TracingQueueErrorDiagnosticsTest.php b/tests/Unit/V2/TracingQueueErrorDiagnosticsTest.php index 9d1b829c..b6e5dfdf 100644 --- a/tests/Unit/V2/TracingQueueErrorDiagnosticsTest.php +++ b/tests/Unit/V2/TracingQueueErrorDiagnosticsTest.php @@ -114,6 +114,62 @@ public static function unsafeRefusals(): array ]; } + public function testOnlyKnownQueueDispatchContentionIsClassified(): void + { + $error = self::queueDispatchRefusal(); + foreach ([5, 6, 517] as $code) { + $error['native_code'] = $code; + self::assertTrue(TracingQueueErrorDiagnostics::isQueueDispatchContention($error)); + } + unset($error['ack_fence']); + self::assertFalse(TracingQueueErrorDiagnostics::isQueueDispatchContention($error)); + } + + #[DataProvider('unsafeQueueDispatchReports')] + public function testOtherQueueDispatchErrorsRemainRefusals(array $changes): void + { + self::assertFalse(TracingQueueErrorDiagnostics::isQueueDispatchContention( + array_replace(self::queueDispatchRefusal(), $changes), + )); + } + + public static function unsafeQueueDispatchReports(): array + { + return self::unsafeRefusals() + [ + 'other exception' => [['error_class' => RuntimeException::class]], + 'untyped sqlite code' => [['native_code' => '5']], + 'invalid negative sqlite code' => [['native_code' => -251]], + 'acknowledgment instead of dispatch' => [['operation' => 'delete']], + 'unexpected acknowledgment fence' => [['ack_fence' => [1, 'default', 10, 1]]], + 'missing producer frame' => [['stage_frames' => [[ + 'file' => 'DatabaseQueue.php', 'function' => 'insertGetId', + ]]]], + 'missing insert frame' => [['stage_frames' => [[ + 'file' => 'DatabaseQueue.php', 'function' => 'pushToDatabase', + ]]]], + ]; + } + + private static function queueDispatchRefusal(): array + { + return [ + 'error_class' => QueryException::class, + 'native_code' => 5, + 'operation' => 'insert', + 'transaction_level' => 0, + 'pdo_in_transaction' => false, + 'ack_fence' => null, + 'statement_truncated' => false, + 'statement' => 'insert into "jobs" ' + . '("queue", "attempts", "reserved_at", "available_at", "created_at", "payload") values (?, ?, ?, ?, ?, ?)', + 'stage_frames' => [[ + 'file' => 'DatabaseQueue.php', 'function' => 'insertGetId', + ], [ + 'file' => 'DatabaseQueue.php', 'function' => 'pushToDatabase', + ]], + ]; + } + private static function failedJobLogRefusal(): array { return [ From d65f025070ddb4aa7e4be02dbcac42f4b547cd63 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Wed, 21 Oct 2026 03:45:46 +0000 Subject: [PATCH 2/2] Apply ECS formatting to queue diagnostic controls --- .../V2/TracingQueueErrorDiagnosticsTest.php | 44 +++++++++++++------ 1 file changed, 31 insertions(+), 13 deletions(-) diff --git a/tests/Unit/V2/TracingQueueErrorDiagnosticsTest.php b/tests/Unit/V2/TracingQueueErrorDiagnosticsTest.php index b6e5dfdf..6f00fa6b 100644 --- a/tests/Unit/V2/TracingQueueErrorDiagnosticsTest.php +++ b/tests/Unit/V2/TracingQueueErrorDiagnosticsTest.php @@ -136,17 +136,33 @@ public function testOtherQueueDispatchErrorsRemainRefusals(array $changes): void public static function unsafeQueueDispatchReports(): array { return self::unsafeRefusals() + [ - 'other exception' => [['error_class' => RuntimeException::class]], - 'untyped sqlite code' => [['native_code' => '5']], - 'invalid negative sqlite code' => [['native_code' => -251]], - 'acknowledgment instead of dispatch' => [['operation' => 'delete']], - 'unexpected acknowledgment fence' => [['ack_fence' => [1, 'default', 10, 1]]], - 'missing producer frame' => [['stage_frames' => [[ - 'file' => 'DatabaseQueue.php', 'function' => 'insertGetId', - ]]]], - 'missing insert frame' => [['stage_frames' => [[ - 'file' => 'DatabaseQueue.php', 'function' => 'pushToDatabase', - ]]]], + 'other exception' => [[ + 'error_class' => RuntimeException::class, + ]], + 'untyped sqlite code' => [[ + 'native_code' => '5', + ]], + 'invalid negative sqlite code' => [[ + 'native_code' => -251, + ]], + 'acknowledgment instead of dispatch' => [[ + 'operation' => 'delete', + ]], + 'unexpected acknowledgment fence' => [[ + 'ack_fence' => [1, 'default', 10, 1], + ]], + 'missing producer frame' => [[ + 'stage_frames' => [[ + 'file' => 'DatabaseQueue.php', + 'function' => 'insertGetId', + ]], + ]], + 'missing insert frame' => [[ + 'stage_frames' => [[ + 'file' => 'DatabaseQueue.php', + 'function' => 'pushToDatabase', + ]], + ]], ]; } @@ -163,9 +179,11 @@ private static function queueDispatchRefusal(): array 'statement' => 'insert into "jobs" ' . '("queue", "attempts", "reserved_at", "available_at", "created_at", "payload") values (?, ?, ?, ?, ?, ?)', 'stage_frames' => [[ - 'file' => 'DatabaseQueue.php', 'function' => 'insertGetId', + 'file' => 'DatabaseQueue.php', + 'function' => 'insertGetId', ], [ - 'file' => 'DatabaseQueue.php', 'function' => 'pushToDatabase', + 'file' => 'DatabaseQueue.php', + 'function' => 'pushToDatabase', ]], ]; }