Retry the documented poll lock-pressure refusal in every parity probe

scripts/conformance/server-parity/activity-deadline-probe.php

@@ -34,7 +34,7 @@ function activityDeadlineHttpObservation(array $fixture, array $options): array
         if (!preg_match('#^http://127\.0\.0\.1:[0-9]+$#', $url)) {
             throw new RuntimeException('Activity deadline recorder did not bind its local port');
         }
-        $client = new Client($url, namespace:'default', token:getenv('DW_PARITY_TOKEN') ?: null);
+        $client = parityClient($url, namespace:'default', token:getenv('DW_PARITY_TOKEN') ?: null);
         $checkBudget = static function () use ($deadline): void {
             if (microtime(true) >= $deadline) {
                 throw new RuntimeException('Original activity deadline observation expired');

scripts/conformance/server-parity/admission-probe.php

@@ -86,7 +86,7 @@ function finishHttpAdmission(Client $client, array $fixture, string $workflowId,
         throw $error;
     } finally {
         if (isset($definition['cleanup_auth'])) {
-            $cleanupClient = new Client($url, namespace: $namespace, token: parityAdmissionToken($definition['cleanup_auth']));
+            $cleanupClient = parityClient($url, namespace: $namespace, token: parityAdmissionToken($definition['cleanup_auth']));
             $state['cleanup_auth'] = $definition['cleanup_auth'];
             $state['cleanup'] = $cleanupClient->cancelWorkflow($peerId, $definition['cleanup_reason'], $peer->selectedRunId);
         } else {

scripts/conformance/server-parity/cancellation-probe.php

@@ -40,7 +40,7 @@ function cancellationReceipt(callable $submit): array
 function httpCancellationObservation(array $fixture, string $workflowId, string $namespace, string $queue, string $url): array
 {
     $polls = $completions = $activityPolls = $activityOutcomes = [];
-    $client = new Client($url, namespace: $namespace, token: getenv('DW_PARITY_TOKEN') ?: null,
+    $client = parityClient($url, namespace: $namespace, token: getenv('DW_PARITY_TOKEN') ?: null,
         transport: observedChildTransport($polls, $completions, $activityPolls, $activityOutcomes));
     $phase = $fixture['immediate_cancellation']['phase'];
     $reason = $fixture['immediate_cancellation']['reason'];
@@ -113,7 +113,7 @@ function httpCancellationObservation(array $fixture, string $workflowId, string
             $task['activity_attempt_id'], $task['lease_owner'], $fixture['input']));
     }
     // A separate published Client reloads the same durable run and history.
-    $fresh = new Client($url, namespace: $namespace, token: getenv('DW_PARITY_TOKEN') ?: null);
+    $fresh = parityClient($url, namespace: $namespace, token: getenv('DW_PARITY_TOKEN') ?: null);
     $inspectionWorker = $workflowId.':inspection';
     $fresh->registerWorker($inspectionWorker, $queue, [$fixture['workflow_type']], ['parity.v1.cancel_activity']);
     $remaining = ['workflow' => $fresh->pollWorkflowTask($inspectionWorker, $queue, 0),

scripts/conformance/server-parity/child-cancellation-probe.php

@@ -54,7 +54,7 @@ function finishHttpChildCancellation(Client $client, string $parentId, string $p
     $state['terminal_duplicate'] = cancellationReceipt(static fn (): array => $client->cancelWorkflow($id, 'post-parent-completion replacement', $runId));
     $state['child_history_after_terminal_duplicate'] = httpHistory($client, $id, $runId);
     $state['parent_history_after_terminal_duplicate'] = httpHistory($client, $parentId, $parentRunId);
-    $fresh = new Client($url, namespace: $namespace, token: getenv('DW_PARITY_TOKEN') ?: null);
+    $fresh = parityClient($url, namespace: $namespace, token: getenv('DW_PARITY_TOKEN') ?: null);
     $state['fresh_child'] = $fresh->describeWorkflow($id, $runId)->raw;
     $state['fresh_child_history'] = httpHistory($fresh, $id, $runId);
     $state['fresh_parent'] = $fresh->describeWorkflow($parentId, $parentRunId)->raw;

scripts/conformance/server-parity/cooperative-probe.php

@@ -75,7 +75,7 @@ function httpCooperativeObservation(array $fixture, string $workflowId, string $
 {
     $traffic = [];
     $diagnostics = [];
-    $client = new Client($url, namespace: $namespace, token: getenv('DW_PARITY_TOKEN') ?: null,
+    $client = parityClient($url, namespace: $namespace, token: getenv('DW_PARITY_TOKEN') ?: null,
         transport: cooperativeObservedTransport($traffic), workerProtocolVersion: '1.20');
     $phase = $fixture['cooperative_cancellation']['phase'];
     $reason = $fixture['cooperative_cancellation']['reason'];
@@ -150,7 +150,7 @@ static function (string $event, array $details) use (&$handle, &$diagnostics, $c
     } catch (WorkflowCancelled $error) {
         $outcome = ['type' => $error::class, 'message' => $error->getMessage()];
     }
-    $fresh = new Client($url, namespace: $namespace, token: getenv('DW_PARITY_TOKEN') ?: null, workerProtocolVersion: '1.20');
+    $fresh = parityClient($url, namespace: $namespace, token: getenv('DW_PARITY_TOKEN') ?: null, workerProtocolVersion: '1.20');
 
     return ['execution' => $execution->raw, 'workflow_id' => $execution->workflowId, 'run_id' => $execution->runId,
         'workflow_type' => $execution->workflowType, 'namespace' => $execution->namespace, 'task_queue' => $execution->taskQueue,

scripts/conformance/server-parity/cooperative-restart.php

@@ -8,6 +8,7 @@
 use DurableWorkflow\Worker\CancellationDelivery;
 
 require __DIR__.'/vendor/autoload.php';
+require_once __DIR__.'/lock-pressure-transport.php';
 require __DIR__.'/observation-values.php';
 require __DIR__.'/cooperative-probe.php';
 
@@ -56,7 +57,7 @@ function cooperativeEvent(array $events, string $type): array
         $workflowId = 'root-restart-'.$outcome;
         $queue = 'root-restart-'.$outcome;
         $traffic = [];
-        $client = new Client($url, namespace: 'default', token: getenv('DW_PARITY_TOKEN') ?: null,
+        $client = parityClient($url, namespace: 'default', token: getenv('DW_PARITY_TOKEN') ?: null,
             transport: cooperativeObservedTransport($traffic), workerProtocolVersion: '1.20');
         $worker = $handle = $accepted = null;
         $guard = microtime(true) + 20;
@@ -118,7 +119,7 @@ static function (string $event) use (&$handle, $client, $fixture, $workflowId, $
 foreach (['deadline_expired', 'completed'] as $outcome) {
     $receipt = $receipts[$outcome];
     $traffic = [];
-    $client = new Client($url, namespace: 'default', token: getenv('DW_PARITY_TOKEN') ?: null,
+    $client = parityClient($url, namespace: 'default', token: getenv('DW_PARITY_TOKEN') ?: null,
         transport: cooperativeObservedTransport($traffic), workerProtocolVersion: '1.20');
     $handle = $client->workflowHandle($receipt['workflow_id'], $receipt['run_id']);
     $before = httpHistory($client, $receipt['workflow_id'], $receipt['run_id']);
@@ -188,7 +189,7 @@ static function () use (&$worker, $handle, $guard): float {
     }
     $duplicate = $client->requestWorkflowCancellation($receipt['workflow_id'], 'post-terminal replacement', 3600, $receipt['run_id']);
     assertCooperativeRecovery($duplicate['duplicate'] === true && $duplicate['cancellation_request']['cleanup_deadline_at'] === $receipt['context']['cleanup_deadline_at'], 'Terminal duplicate changed the original budget.');
-    $fresh = new Client($url, namespace: 'default', token: getenv('DW_PARITY_TOKEN') ?: null, workerProtocolVersion: '1.20');
+    $fresh = parityClient($url, namespace: 'default', token: getenv('DW_PARITY_TOKEN') ?: null, workerProtocolVersion: '1.20');
     assertCooperativeRecovery(sameCooperativeValue($events, httpHistory($fresh, $receipt['workflow_id'], $receipt['run_id'])), 'Late calls or fresh reads changed terminal history.');
     try {
         $fresh->workflowHandle($receipt['workflow_id'], $receipt['run_id'])->resultOfSelectedRun(1);

scripts/conformance/server-parity/lock-pressure-transport.php

@@ -44,3 +44,27 @@ public function send(string $method, string $uri, array $headers, ?array $body =
         }
     }
 }
+
+/**
+ * Creates an SDK client for a parity probe with the documented poll retry
+ * applied to its transport, keeping any transport the caller supplies. Any
+ * retries are reported once on stderr when the process exits, so recorded
+ * observations stay comparable across targets.
+ */
+function parityClient(mixed ...$arguments): \DurableWorkflow\Client
+{
+    static $transports = [];
+    $transport = new LockPressurePollRetryTransport($arguments['transport'] ?? null);
+    if ($transports === []) {
+        register_shutdown_function(static function () use (&$transports): void {
+            $retries = array_merge(...array_map(static fn (LockPressurePollRetryTransport $used): array => $used->retries, $transports));
+            if ($retries !== []) {
+                fwrite(STDERR, json_encode(['lock_pressure_poll_retries' => $retries], JSON_THROW_ON_ERROR).PHP_EOL);
+            }
+        });
+    }
+    $transports[] = $transport;
+    $arguments['transport'] = $transport;
+
+    return new \DurableWorkflow\Client(...$arguments);
+}

scripts/conformance/server-parity/namespace-probe.php

@@ -98,7 +98,7 @@ function finishHttpNamespaces(Client $admin, array $fixture, string $workflowId,
             'before' => $before, 'after' => namespaceHttpRun($clients[0], $ids[0], $handles[0]->selectedRunId)];
     };
     foreach ($names as $index => $name) {
-        $clients[] = new Client($url, namespace: $name, token: getenv('DW_PARITY_TOKEN') ?: null,
+        $clients[] = parityClient($url, namespace: $name, token: getenv('DW_PARITY_TOKEN') ?: null,
             transport: namespaceTransport($state['worker_receipts'][$index], $index === 0 ? $beforeCompletion : null));
         $state['registrations'][] = $clients[$index]->registerWorker($state['worker_id'], $queue,
             [$definition['peer_workflow_type']], ['parity.v1.echo_activity']);

scripts/conformance/server-parity/patch-deployment-probe.php

@@ -126,7 +126,7 @@ function patchHttpPhase(array $fixture, array $options): array
             });
         };
     });
-    $client = new Client($options['url'], namespace: 'default', token: getenv('DW_PARITY_TOKEN') ?: null,
+    $client = parityClient($options['url'], namespace: 'default', token: getenv('DW_PARITY_TOKEN') ?: null,
         transport: new Psr18Transport(client: new GuzzleClient(['http_errors' => false, 'handler' => $stack])));
     if ($phase === 'replacement') {
         $handle = $client->workflowHandle($workflowId, $options['run-id']);
@@ -279,7 +279,7 @@ function patchPublishedSdkPhase(array $fixture, array $options): array
         if ($result['language'] !== $consumer['language'] || $result['sdk_version'] !== $consumer['version']) {
             throw new RuntimeException('Actual published worker differs from the selected consumer.');
         }
-        $client = new Client($options['url'], namespace: 'default', token: getenv('DW_PARITY_TOKEN') ?: null);
+        $client = parityClient($options['url'], namespace: 'default', token: getenv('DW_PARITY_TOKEN') ?: null);
         $runId = $original ? ($result['run_id'] ?? null) : $options['run-id'];
         if (! is_string($runId) || $runId === '') {
             throw new RuntimeException('Actual published author returned no original run identity.');

scripts/conformance/server-parity/probe.php

@@ -64,6 +64,9 @@
 
 require __DIR__.'/observation-values.php';
 require __DIR__.'/probe-failure.php';
+// A published PHP SQLite reference shares its file with its maintenance loop
+// and may refuse a worker poll with the documented backend_lock_pressure reply.
+require_once __DIR__.'/lock-pressure-transport.php';
 
 function observedChildTransport(array &$polls, array &$completions, array &$activityPolls, array &$activityOutcomes, ?Closure $afterCompletion = null, ?array &$scheduleReceipts = null): Psr18Transport
 {
@@ -114,7 +117,7 @@ function observedChildTransport(array &$polls, array &$completions, array &$acti
 // client in an independent PHP process so this probe's worker keeps polling.
 if ($mode === 'query') {
     try {
-        $client = new Client($options['url'], namespace: $namespace, token: getenv('DW_PARITY_TOKEN') ?: null);
+        $client = parityClient($options['url'], namespace: $namespace, token: getenv('DW_PARITY_TOKEN') ?: null);
         $result = $client->queryWorkflow($workflowId, $fixture['query_name'], $fixture['query_arguments'], $options['run-id']);
         echo json_encode(['result' => $result, 'typed_result' => typedValue($result)], JSON_THROW_ON_ERROR | JSON_UNESCAPED_UNICODE)."\n";
         exit(0);
@@ -142,9 +145,9 @@ function httpObservation(array $fixture, string $workflowId, string $namespace,
     } : null;
     $transport = isset($fixture['child_count']) || isset($fixture['child_cancellation']) || isset($fixture['retry_policy']) || isset($fixture['schedule'])
         ? observedChildTransport($workflowPolls, $workflowCompletions, $activityPolls, $activityOutcomes, $afterCompletion, $scheduleReceipts) : null;
-    $client = new Client($url, namespace: $namespace, transport: $transport, token: getenv('DW_PARITY_TOKEN') ?: null);
+    $client = parityClient($url, namespace: $namespace, transport: $transport, token: getenv('DW_PARITY_TOKEN') ?: null);
     $workerClient = ($fixture['auth_profile'] ?? null) === 'role_tokens'
-        ? new Client($url, namespace: $namespace, token: parityAdmissionToken('worker')) : $client;
+        ? parityClient($url, namespace: $namespace, token: parityAdmissionToken('worker')) : $client;
     $handle = null;
     $deliveries = [];
     $queries = [];

scripts/conformance/server-parity/sdk-direction-probe.php

@@ -73,15 +73,15 @@ function sdkDirectionHttpObservation(array $fixture, array $options, ?Closure $a
     $spec = $fixture[$directionKey];
     $childWorkflowId = $childRunId = '';
     $namespace = $fixture['namespace'] ?? 'default';
-    $client = new Client($options['url'], namespace: $namespace, token: getenv('DW_PARITY_TOKEN') ?: null);
+    $client = parityClient($options['url'], namespace: $namespace, token: getenv('DW_PARITY_TOKEN') ?: null);
     $deadline = $scenarioDeadline ?? microtime(true) + 35;
     $observerStart = null;
     try {
         $url = trim(fgets($proxyPipes[1]));
         if (! preg_match('#^http://127\\.0\\.0\\.1:[0-9]+$#', $url)) {
             throw new RuntimeException('SDK observer did not bind its local port.');
         }
-        $startClient = $signalDirection ? new Client($url, namespace: $namespace, token: getenv('DW_PARITY_TOKEN') ?: null) : null;
+        $startClient = $signalDirection ? parityClient($url, namespace: $namespace, token: getenv('DW_PARITY_TOKEN') ?: null) : null;
         foreach ($spec['phase_order'] as $phase) {
             $language = $signalDirection ? $spec[in_array($phase, ['signal-1', 'signal-2'], true) ? 'client' : 'workflow']
                 : ($timerAuthor ? $spec['language'] : ($childDirection ? $spec[$phase === 'child' ? 'child' : 'parent']

scripts/conformance/server-parity/sdk-namespace-signal-probe.php

@@ -158,7 +158,7 @@ function namespaceSignalHttpObservation(array $fixture, array $options): array
             throw new RuntimeException('Namespace control observer did not bind its local port.');
         }
         $checkpoint('client_setup');
-        $admin = new Client($url, namespace: 'default', token: getenv('DW_PARITY_TOKEN') ?: null);
+        $admin = parityClient($url, namespace: 'default', token: getenv('DW_PARITY_TOKEN') ?: null);
         foreach ($fixtures as $index => $runFixture) {
             try {
                 $checkpoint('namespace_describe', $runFixture['namespace']);
@@ -175,7 +175,7 @@ function namespaceSignalHttpObservation(array $fixture, array $options): array
             $state['administration'][] = ['created' => $created, 'namespace' => $namespace->raw];
             $paths[$index] = tempnam(sys_get_temp_dir(), 'parity-namespace-fixture-');
             file_put_contents($paths[$index], json_encode($runFixture, JSON_THROW_ON_ERROR | JSON_UNESCAPED_UNICODE));
-            $clients[$index] = new Client($options['url'], namespace: $runFixture['namespace'], token: getenv('DW_PARITY_TOKEN') ?: null);
+            $clients[$index] = parityClient($options['url'], namespace: $runFixture['namespace'], token: getenv('DW_PARITY_TOKEN') ?: null);
             $start = microtime(true);
             $deadline = $start + 35;
             $state['run_budgets'][$index] = ['started_at_unix' => $start, 'deadline_at_unix' => $deadline];

scripts/conformance/server-parity/sdk-query-direction-probe.php

@@ -242,7 +242,7 @@ function sdkQueryDirectionHttpObservation(array $fixture, array $options): array
         throw new RuntimeException('SDK query direction requires its evidence directory.');
     }
     $spec = $fixture['sdk_query_direction'];
-    $client = new Client($options['url'], namespace: 'default', token: getenv('DW_PARITY_TOKEN') ?: null);
+    $client = parityClient($options['url'], namespace: 'default', token: getenv('DW_PARITY_TOKEN') ?: null);
     $deadline = microtime(true) + 35;
     $runId = '';
     $phases = $actors = [];
@@ -262,7 +262,7 @@ function sdkQueryDirectionHttpObservation(array $fixture, array $options): array
                         if ($request['method'] === 'POST' && $request['path'] === '/api/worker/register'
                             && $request['status'] === 201 && ($request['response']['registered'] ?? null) === true
                             && ($request['request']['worker_id'] ?? null) === $options['workflow-id'].':original') {
-                            $startClient = new Client($author['url'], namespace: 'default', token: getenv('DW_PARITY_TOKEN') ?: null);
+                            $startClient = parityClient($author['url'], namespace: 'default', token: getenv('DW_PARITY_TOKEN') ?: null);
                             $runId = $startClient->startWorkflow($fixture['workflow_type'], $options['workflow-id'],
                                 $spec['task_queue'], [$fixture['input'], $fixture['signal_count']])->selectedRunId;
                             $observerStart = ['role' => 'observer', 'language' => 'php', 'pid' => getmypid(),

scripts/conformance/server-parity/sdk-selected-run-signal-probe.php

@@ -70,7 +70,7 @@ function selectedRunSignalHttpObservation(array $fixture, array $options): array
         if (! preg_match('#^http://127\.0\.0\.1:[0-9]+$#', $url)) {
             throw new RuntimeException('Selected-run control observer failed to bind.');
         }
-        $admin = new Client($url, namespace: 'default', token: getenv('DW_PARITY_TOKEN') ?: null);
+        $admin = parityClient($url, namespace: 'default', token: getenv('DW_PARITY_TOKEN') ?: null);
         foreach ($fixtures as $index => $runFixture) {
             $checkpoint('namespace_setup_'.$index);
             try {
@@ -84,7 +84,7 @@ function selectedRunSignalHttpObservation(array $fixture, array $options): array
             $state['administration'][] = ['created' => $created, 'namespace' => $namespace->raw];
             $paths[$index] = tempnam(sys_get_temp_dir(), 'parity-selected-run-');
             file_put_contents($paths[$index], json_encode($runFixture, JSON_THROW_ON_ERROR | JSON_UNESCAPED_UNICODE));
-            $clients[$index] = new Client($options['url'], namespace: $runFixture['namespace'], token: getenv('DW_PARITY_TOKEN') ?: null);
+            $clients[$index] = parityClient($options['url'], namespace: $runFixture['namespace'], token: getenv('DW_PARITY_TOKEN') ?: null);
             $start = microtime(true);
             $deadline = $start + 35;
             $state['run_budgets'][$index] = ['started_at_unix' => $start, 'deadline_at_unix' => $deadline];

scripts/conformance/server-parity/sdk-update-direction-probe.php

@@ -78,7 +78,7 @@ function sdkUpdateDirectionHttpObservation(array $fixture, array $options): arra
         throw new RuntimeException('SDK update direction requires its evidence directory.');
     }
     $spec = $fixture['sdk_update_direction'];
-    $client = new Client($options['url'], namespace: 'default', token: getenv('DW_PARITY_TOKEN') ?: null);
+    $client = parityClient($options['url'], namespace: 'default', token: getenv('DW_PARITY_TOKEN') ?: null);
     $deadline = microtime(true) + 35;
     $workflowId = $options['workflow-id'];
     $runId = '';
@@ -113,7 +113,7 @@ function sdkUpdateDirectionHttpObservation(array $fixture, array $options): arra
                     if ($request['method'] === 'POST' && $request['path'] === '/api/worker/register'
                         && $request['status'] === 201 && ($request['response']['registered'] ?? null) === true
                         && ($request['request']['worker_id'] ?? null) === $workflowId.':original') {
-                        $start = new Client($actor['url'], namespace: 'default', token: getenv('DW_PARITY_TOKEN') ?: null);
+                        $start = parityClient($actor['url'], namespace: 'default', token: getenv('DW_PARITY_TOKEN') ?: null);
                         $runId = $start->startWorkflow($fixture['workflow_type'], $workflowId,
                             $spec['task_queue'], [$fixture['input'], $fixture['signal_count']])->selectedRunId;
                         $observerStart = ['role' => 'observer', 'language' => 'php', 'pid' => getmypid(),