From 0ac1b8feacb57b38b7cbfdfa08cf0c8e8e92b3bd Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 07:56:11 +0000 Subject: [PATCH 1/5] Retry the PHP reference's lock-pressure refusals below the SDK transport The published PHP SQLite reference shares its database file with its maintenance loop and sometimes refuses a request with its documented HTTP 503 backend_lock_pressure reply: retryable, nothing committed, worker mutations fenced by task and attempt. Probe runs failed on these refusals for polls, worker operations and a Rust SDK actor's deregistration. - Parity probes now build their SDK clients through parityClient(), whose default transport is the SDK's own Psr18Transport over Guzzle with an innermost retry middleware. The transport keeps every capability the SDK checks (bounded requests and payload transfer), unlike the earlier wrapper. Observed-transport stacks add the same middleware innermost, so recording middleware sees only the final reply. - lockPressureOperationRetryDelay retries any non-poll request only on status 503, reason backend_lock_pressure and retryable true, within 15 retries, 30 s of waiting and the request's own timeout. Polls keep the stricter rule that no task was handed out. Bodies that cannot be replayed are never retried. Retries are reported once on stderr. - The Rust SDK actor retries its deregistration on the same refusal, under the same bounds. Co-Authored-By: Claude Opus 5.5 --- .../server-parity/activity-deadline-probe.php | 3 +- .../server-parity/admission-probe.php | 3 +- .../server-parity/cancellation-probe.php | 5 +- .../child-cancellation-probe.php | 3 +- .../server-parity/cooperative-probe.php | 6 +- .../server-parity/cooperative-restart.php | 7 +- .../rust/src/lock_pressure.rs | 94 ++++++++++++++++++ .../corrected-markers/rust/src/main.rs | 6 +- .../corrected-markers/rust/src/memo_author.rs | 5 +- .../lock-pressure-middleware.php | 95 +++++++++++++++++++ .../server-parity/lock-pressure-retry.php | 42 ++++++++ .../server-parity/namespace-probe.php | 4 +- .../server-parity/patch-deployment-probe.php | 6 +- scripts/conformance/server-parity/probe.php | 8 +- .../server-parity/sdk-direction-probe.php | 5 +- .../sdk-namespace-signal-probe.php | 5 +- .../sdk-query-direction-probe.php | 5 +- .../sdk-selected-run-signal-probe.php | 5 +- .../sdk-update-direction-probe.php | 5 +- .../server-parity/task-retry-probe.php | 3 +- .../server-parity/waiting-history-probe.php | 3 +- .../ServerParityLockPressureRetryTest.php | 24 +++++ 22 files changed, 312 insertions(+), 30 deletions(-) create mode 100644 scripts/conformance/server-parity/corrected-markers/rust/src/lock_pressure.rs create mode 100644 scripts/conformance/server-parity/lock-pressure-middleware.php diff --git a/scripts/conformance/server-parity/activity-deadline-probe.php b/scripts/conformance/server-parity/activity-deadline-probe.php index b4bd61dfe..4728f6264 100644 --- a/scripts/conformance/server-parity/activity-deadline-probe.php +++ b/scripts/conformance/server-parity/activity-deadline-probe.php @@ -3,6 +3,7 @@ declare(strict_types=1); use DurableWorkflow\Client; +require_once __DIR__.'/lock-pressure-middleware.php'; require_once __DIR__.'/namespace-probe.php'; require_once __DIR__.'/activity-deadline-poll.php'; @@ -34,7 +35,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'); diff --git a/scripts/conformance/server-parity/admission-probe.php b/scripts/conformance/server-parity/admission-probe.php index 0e681a1db..337bd9564 100644 --- a/scripts/conformance/server-parity/admission-probe.php +++ b/scripts/conformance/server-parity/admission-probe.php @@ -8,6 +8,7 @@ use Workflow\V2\StartOptions; use Workflow\V2\WorkflowStub; use Workflow\WorkflowOptions; +require_once __DIR__.'/lock-pressure-middleware.php'; function parityAdmissionToken(string $auth): string { @@ -86,7 +87,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 { diff --git a/scripts/conformance/server-parity/cancellation-probe.php b/scripts/conformance/server-parity/cancellation-probe.php index 0292c6abc..98f31d054 100644 --- a/scripts/conformance/server-parity/cancellation-probe.php +++ b/scripts/conformance/server-parity/cancellation-probe.php @@ -17,6 +17,7 @@ use Workflow\V2\WorkflowStub; use Workflow\V2\StartOptions; use Workflow\WorkflowOptions; +require_once __DIR__.'/lock-pressure-middleware.php'; // Application callbacks use only the published APIs. These observations never // mutate engine rows or synthesize a claim, cancellation or terminal outcome. @@ -40,7 +41,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 +114,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), diff --git a/scripts/conformance/server-parity/child-cancellation-probe.php b/scripts/conformance/server-parity/child-cancellation-probe.php index b548be1bb..7b8ee4449 100644 --- a/scripts/conformance/server-parity/child-cancellation-probe.php +++ b/scripts/conformance/server-parity/child-cancellation-probe.php @@ -5,6 +5,7 @@ use DurableWorkflow\Client; use Workflow\V2\Models\WorkflowRun; use Workflow\V2\WorkflowStub; +require_once __DIR__.'/lock-pressure-middleware.php'; final class ChildCancellationProbeState { @@ -54,7 +55,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; diff --git a/scripts/conformance/server-parity/cooperative-probe.php b/scripts/conformance/server-parity/cooperative-probe.php index cb2e19d83..b856b308c 100644 --- a/scripts/conformance/server-parity/cooperative-probe.php +++ b/scripts/conformance/server-parity/cooperative-probe.php @@ -21,10 +21,12 @@ use Workflow\V2\WorkflowStub; use Workflow\V2\StartOptions; use Workflow\WorkflowOptions; +require_once __DIR__.'/lock-pressure-middleware.php'; function cooperativeObservedTransport(array &$traffic): Psr18Transport { $stack = HandlerStack::create(); + $stack->unshift(LockPressurePollRetries::middleware(), 'lock_pressure_poll_retry'); $stack->push(static function (callable $handler) use (&$traffic): callable { return static function (RequestInterface $request, array $options) use ($handler, &$traffic) { return $handler($request, $options)->then(static function (ResponseInterface $response) use ($request, &$traffic): ResponseInterface { @@ -75,7 +77,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 +152,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, diff --git a/scripts/conformance/server-parity/cooperative-restart.php b/scripts/conformance/server-parity/cooperative-restart.php index 6fd9950b3..2e4b81d3b 100644 --- a/scripts/conformance/server-parity/cooperative-restart.php +++ b/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-middleware.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); diff --git a/scripts/conformance/server-parity/corrected-markers/rust/src/lock_pressure.rs b/scripts/conformance/server-parity/corrected-markers/rust/src/lock_pressure.rs new file mode 100644 index 000000000..5f9f9f9b9 --- /dev/null +++ b/scripts/conformance/server-parity/corrected-markers/rust/src/lock_pressure.rs @@ -0,0 +1,94 @@ +//! The published PHP SQLite reference can refuse a worker operation with its +//! documented HTTP 503 `backend_lock_pressure` reply and `retryable: true` +//! after its transaction failed, so nothing was committed. Only that refusal is +//! retried, within 15 retries and 30 seconds of waiting, as the PHP probes' +//! transport does (`../lock-pressure-retry.php`). Every other outcome is +//! returned unchanged. +use durable_workflow::{Error, Result}; +use serde_json::Value; +use std::{future::Future, time::Duration}; + +const MAX_RETRIES: u32 = 15; +const MAX_WAIT: Duration = Duration::from_secs(30); + +fn delay_for(status: u16, body: &str, retries: u32, waited: Duration) -> Option { + if status != 503 || retries >= MAX_RETRIES { + return None; + } + let body: Value = serde_json::from_str(body).ok()?; + if body.get("reason")?.as_str()? != "backend_lock_pressure" + || body.get("retryable") != Some(&Value::Bool(true)) + { + return None; + } + let seconds = match body.get("retry_after_seconds") { + None => 1.0, + Some(value) => value.as_f64()?, + }; + let delay = Duration::from_secs_f64(seconds.clamp(0.1, 5.0)); + (waited + delay <= MAX_WAIT).then_some(delay) +} + +/// Runs `call`, repeating it only after the documented lock-pressure refusal. +pub async fn retrying(what: &str, mut call: F) -> Result +where + F: FnMut() -> Fut, + Fut: Future>, +{ + let (mut retries, mut waited) = (0, Duration::ZERO); + loop { + let error = match call().await { + Err(error) => error, + done => return done, + }; + let delay = match &error { + Error::Http { status, body } => delay_for(status.as_u16(), body, retries, waited), + _ => None, + }; + let Some(delay) = delay else { + return Err(error); + }; + eprintln!("{what}: retrying backend_lock_pressure refusal after {delay:?}"); + tokio::time::sleep(delay).await; + waited += delay; + retries += 1; + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const REFUSAL: &str = r#"{"reason":"backend_lock_pressure","operation":"destroy","outcome":"deferred","retryable":true,"retry_after_seconds":1}"#; + + #[test] + fn documented_refusal_waits_for_the_advertised_interval() { + assert_eq!(delay_for(503, REFUSAL, 0, Duration::ZERO), Some(Duration::from_secs(1))); + let long = REFUSAL.replace(r#""retry_after_seconds":1"#, r#""retry_after_seconds":60"#); + assert_eq!(delay_for(503, &long, 0, Duration::ZERO), Some(Duration::from_secs(5))); + let absent = REFUSAL.replace(r#","retry_after_seconds":1"#, ""); + assert_eq!(delay_for(503, &absent, 0, Duration::ZERO), Some(Duration::from_secs(1))); + } + + #[test] + fn other_statuses_reasons_and_bodies_are_returned() { + for (status, body) in [ + (409, REFUSAL.to_owned()), + (500, REFUSAL.to_owned()), + (503, REFUSAL.replace("backend_lock_pressure", "backend_unavailable")), + (503, REFUSAL.replace(r#""retryable":true"#, r#""retryable":false"#)), + (503, REFUSAL.replace(r#","retryable":true"#, "")), + (503, REFUSAL.replace(r#""retry_after_seconds":1"#, r#""retry_after_seconds":"soon""#)), + (503, "not json".to_owned()), + ] { + assert_eq!(delay_for(status, &body, 0, Duration::ZERO), None, "{status} {body}"); + } + } + + #[test] + fn retries_stop_at_the_attempt_and_wait_budgets() { + assert_eq!(delay_for(503, REFUSAL, 15, Duration::ZERO), None); + assert_eq!(delay_for(503, REFUSAL, 2, Duration::from_millis(29_500)), None); + assert_eq!(delay_for(503, REFUSAL, 14, Duration::from_secs(29)), Some(Duration::from_secs(1))); + } +} diff --git a/scripts/conformance/server-parity/corrected-markers/rust/src/main.rs b/scripts/conformance/server-parity/corrected-markers/rust/src/main.rs index c5f7bdb7d..3a1d863c7 100644 --- a/scripts/conformance/server-parity/corrected-markers/rust/src/main.rs +++ b/scripts/conformance/server-parity/corrected-markers/rust/src/main.rs @@ -11,6 +11,7 @@ use durable_workflow::{ use serde_json::{json, Value}; mod legacy_range; +mod lock_pressure; mod memo_author; const WORKFLOW: &str = "parity.v1.corrected_markers"; @@ -221,7 +222,10 @@ async fn main() -> Result<()> { } }) .await; - let deregistered = client.deregister_worker_registration(&worker_id).await; + let deregistered = lock_pressure::retrying("deregister", || { + client.deregister_worker_registration(&worker_id) + }) + .await; let result = match outcome { Ok(result) => result, Err(_) => Err(Error::WorkerLoop( diff --git a/scripts/conformance/server-parity/corrected-markers/rust/src/memo_author.rs b/scripts/conformance/server-parity/corrected-markers/rust/src/memo_author.rs index 8520ac901..d2c4c04c2 100644 --- a/scripts/conformance/server-parity/corrected-markers/rust/src/memo_author.rs +++ b/scripts/conformance/server-parity/corrected-markers/rust/src/memo_author.rs @@ -90,7 +90,10 @@ pub async fn run(args: &[String]) -> Result<()> { }) .await .map_err(|_| Error::WorkerLoop("Original memo phase observation expired".into()))?; - let retired = client.deregister_worker_registration(&worker_id).await; + let retired = super::lock_pressure::retrying("deregister", || { + client.deregister_worker_registration(&worker_id) + }) + .await; outcome?; retired?; println!( diff --git a/scripts/conformance/server-parity/lock-pressure-middleware.php b/scripts/conformance/server-parity/lock-pressure-middleware.php new file mode 100644 index 000000000..11cf1d5ef --- /dev/null +++ b/scripts/conformance/server-parity/lock-pressure-middleware.php @@ -0,0 +1,95 @@ + */ + public static array $retries = []; + private static bool $reporting = false; + + public static function middleware(): callable + { + if (!self::$reporting) { + self::$reporting = true; + register_shutdown_function(static function (): void { + if (self::$retries !== []) { + fwrite(STDERR, json_encode(['lock_pressure_poll_retries' => self::$retries], JSON_THROW_ON_ERROR).PHP_EOL); + } + }); + } + + return static fn (callable $handler): callable => static function (RequestInterface $request, array $options) use ($handler) { + $started = microtime(true); + $budget = (float) ($options['timeout'] ?? 0); + $waited = 0.0; + $count = 0; + while (true) { + try { + $response = $handler($request, $options)->wait(); + } catch (\Throwable $error) { + return Create::rejectionFor($error); + } + // A request body that cannot be replayed exactly is never retried. + if (!$response instanceof ResponseInterface || $response->getStatusCode() !== 503 + || (!$request->getBody()->isSeekable() && $request->getBody()->getSize() !== 0)) { + return Create::promiseFor($response); + } + $text = (string) $response->getBody(); + $response = $response->withBody(Utils::streamFor($text)); + $decoded = json_decode($text, true); + $delay = lockPressureOperationRetryDelay($request->getMethod(), (string) $request->getUri(), 503, + is_array($decoded) ? $decoded : null, $count, $waited); + if ($delay === null || ($budget > 0 && microtime(true) - $started + $delay >= $budget)) { + return Create::promiseFor($response); + } + self::$retries[] = ['path' => $request->getUri()->getPath(), 'retry_after_seconds' => $delay]; + usleep((int) round($delay * 1_000_000)); + $waited += $delay; + $count++; + if ($request->getBody()->isSeekable()) { + $request->getBody()->rewind(); + } + } + }; + } + + /** The SDK's default transport, with this retry innermost. */ + public static function transport(): Psr18Transport + { + $stack = HandlerStack::create(); + $stack->unshift(self::middleware(), 'lock_pressure_poll_retry'); + + return new Psr18Transport(client: new GuzzleClient(['http_errors' => false, 'handler' => $stack])); + } +} + +/** An SDK client for a parity probe. A caller-supplied transport is used unchanged. */ +function parityClient(mixed ...$arguments): Client +{ + $arguments['transport'] ??= LockPressurePollRetries::transport(); + + return new Client(...$arguments); +} diff --git a/scripts/conformance/server-parity/lock-pressure-retry.php b/scripts/conformance/server-parity/lock-pressure-retry.php index fa7c28af4..112299d78 100644 --- a/scripts/conformance/server-parity/lock-pressure-retry.php +++ b/scripts/conformance/server-parity/lock-pressure-retry.php @@ -38,3 +38,45 @@ function lockPressurePollRetryDelay( return $waitedSeconds + $delay <= $maxWaitSeconds ? $delay : null; } + +/** + * Decide whether any other request refused for backend lock pressure may be + * retried. + * + * Server returns this refusal as HTTP 503 with `reason: backend_lock_pressure` + * and `retryable: true` only after its transaction failed with a concurrency + * error, so nothing was committed. Worker mutations remain fenced by their + * task and attempt identifiers (`outcome: deferred`), so repeating the request + * cannot record a second outcome. Polls keep lockPressurePollRetryDelay's + * stricter check that no task was handed out. + * + * Returns the delay in seconds before the next attempt, or null to return the + * refusal to the caller unchanged. + */ +function lockPressureOperationRetryDelay( + string $method, + string $uri, + ?int $status, + mixed $response, + int $retriesSoFar, + float $waitedSeconds, + int $maxRetries = 15, + float $maxWaitSeconds = 30.0, +): ?float { + $path = parse_url($uri, PHP_URL_PATH); + if (! is_string($path) || str_ends_with($path, '/poll')) { + return lockPressurePollRetryDelay($method, $uri, $status, $response, $retriesSoFar, $waitedSeconds, + $maxRetries, $maxWaitSeconds); + } + if ($status !== 503 || ! is_array($response) || ($response['reason'] ?? null) !== 'backend_lock_pressure' + || ($response['retryable'] ?? null) !== true || $retriesSoFar >= $maxRetries) { + return null; + } + $retryAfter = $response['retry_after_seconds'] ?? 1; + if (! is_int($retryAfter) && ! is_float($retryAfter)) { + return null; + } + $delay = min(max((float) $retryAfter, 0.1), 5.0); + + return $waitedSeconds + $delay <= $maxWaitSeconds ? $delay : null; +} diff --git a/scripts/conformance/server-parity/namespace-probe.php b/scripts/conformance/server-parity/namespace-probe.php index 7422ec260..4f756671d 100644 --- a/scripts/conformance/server-parity/namespace-probe.php +++ b/scripts/conformance/server-parity/namespace-probe.php @@ -20,6 +20,7 @@ use Workflow\V2\WorkflowStub; use Workflow\WorkflowOptions; use Workflow\Serializers\Serializer; +require_once __DIR__.'/lock-pressure-middleware.php'; function namespaceReceipt(string $id, callable $request): array { @@ -45,6 +46,7 @@ function namespaceHttpRun(Client $client, string $workflowId, string $runId): ar function namespaceTransport(array &$receipts, ?Closure $beforeCompletion = null): Psr18Transport { $stack = HandlerStack::create(); + $stack->unshift(LockPressurePollRetries::middleware(), 'lock_pressure_poll_retry'); $stack->push(static function (callable $handler) use (&$receipts, $beforeCompletion): callable { return static function (RequestInterface $request, array $options) use ($handler, &$receipts, $beforeCompletion) { $path = $request->getUri()->getPath(); @@ -98,7 +100,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']); diff --git a/scripts/conformance/server-parity/patch-deployment-probe.php b/scripts/conformance/server-parity/patch-deployment-probe.php index 032944130..204b073c4 100644 --- a/scripts/conformance/server-parity/patch-deployment-probe.php +++ b/scripts/conformance/server-parity/patch-deployment-probe.php @@ -20,6 +20,7 @@ use Workflow\V2\StartOptions; use Workflow\V2\WorkflowStub; use Workflow\WorkflowOptions; +require_once __DIR__.'/lock-pressure-middleware.php'; final class PatchDeploymentState { @@ -100,6 +101,7 @@ function patchHttpPhase(array $fixture, array $options): array $worker = $handle = null; $requests = $diagnostics = []; $stack = HandlerStack::create(); + $stack->unshift(LockPressurePollRetries::middleware(), 'lock_pressure_poll_retry'); $stack->push(static function (callable $handler) use (&$worker, &$requests, $phase, $fixture): callable { return static function (RequestInterface $request, array $transportOptions) use ($handler, &$worker, &$requests, $phase, $fixture) { return $handler($request, $transportOptions)->then(static function (ResponseInterface $response) use ($request, &$worker, &$requests, $phase, $fixture): ResponseInterface { @@ -126,7 +128,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 +281,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.'); diff --git a/scripts/conformance/server-parity/probe.php b/scripts/conformance/server-parity/probe.php index d63bc0f8d..9c078e331 100644 --- a/scripts/conformance/server-parity/probe.php +++ b/scripts/conformance/server-parity/probe.php @@ -64,12 +64,14 @@ require __DIR__.'/observation-values.php'; require __DIR__.'/probe-failure.php'; +require_once __DIR__.'/lock-pressure-middleware.php'; function observedChildTransport(array &$polls, array &$completions, array &$activityPolls, array &$activityOutcomes, ?Closure $afterCompletion = null, ?array &$scheduleReceipts = null): Psr18Transport { // Observe real published transport I/O. No response or request is changed, // and authentication headers are never retained. $stack = HandlerStack::create(); + $stack->unshift(LockPressurePollRetries::middleware(), 'lock_pressure_poll_retry'); $stack->push(static function (callable $handler) use (&$polls, &$completions, &$activityPolls, &$activityOutcomes, $afterCompletion, &$scheduleReceipts): callable { return static function (RequestInterface $request, array $options) use ($handler, &$polls, &$completions, &$activityPolls, &$activityOutcomes, $afterCompletion, &$scheduleReceipts) { return $handler($request, $options)->then(static function (ResponseInterface $response) use ($request, &$polls, &$completions, &$activityPolls, &$activityOutcomes, $afterCompletion, &$scheduleReceipts): ResponseInterface { @@ -114,7 +116,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 +144,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 = []; diff --git a/scripts/conformance/server-parity/sdk-direction-probe.php b/scripts/conformance/server-parity/sdk-direction-probe.php index b7fa2df5d..1a582e4a8 100644 --- a/scripts/conformance/server-parity/sdk-direction-probe.php +++ b/scripts/conformance/server-parity/sdk-direction-probe.php @@ -5,6 +5,7 @@ use DurableWorkflow\Client; use DurableWorkflow\Worker; use DurableWorkflow\Worker\WorkflowContext; +require_once __DIR__.'/lock-pressure-middleware.php'; function sdkDirectionResponseSummary(mixed $response, bool $workflowHistory = false): mixed { @@ -73,7 +74,7 @@ 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 { @@ -81,7 +82,7 @@ function sdkDirectionHttpObservation(array $fixture, array $options, ?Closure $a 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'] diff --git a/scripts/conformance/server-parity/sdk-namespace-signal-probe.php b/scripts/conformance/server-parity/sdk-namespace-signal-probe.php index 082f7364a..1f7e411d5 100644 --- a/scripts/conformance/server-parity/sdk-namespace-signal-probe.php +++ b/scripts/conformance/server-parity/sdk-namespace-signal-probe.php @@ -11,6 +11,7 @@ use Workflow\V2\StartOptions; use Workflow\V2\WorkflowStub; use Workflow\WorkflowOptions; +require_once __DIR__.'/lock-pressure-middleware.php'; function namespaceSignalRunFixture(array $fixture, bool $peer): array { @@ -158,7 +159,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 +176,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]; diff --git a/scripts/conformance/server-parity/sdk-query-direction-probe.php b/scripts/conformance/server-parity/sdk-query-direction-probe.php index 18dce9429..1beb8c232 100644 --- a/scripts/conformance/server-parity/sdk-query-direction-probe.php +++ b/scripts/conformance/server-parity/sdk-query-direction-probe.php @@ -4,6 +4,7 @@ use Composer\InstalledVersions; use DurableWorkflow\Client; +require_once __DIR__.'/lock-pressure-middleware.php'; function sdkQueryActorOpen(string $language, string $phase, array $fixture, array $options, string $runId, float $deadline): array { @@ -242,7 +243,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 +263,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(), diff --git a/scripts/conformance/server-parity/sdk-selected-run-signal-probe.php b/scripts/conformance/server-parity/sdk-selected-run-signal-probe.php index a97a41a0f..cd11ce1c7 100644 --- a/scripts/conformance/server-parity/sdk-selected-run-signal-probe.php +++ b/scripts/conformance/server-parity/sdk-selected-run-signal-probe.php @@ -10,6 +10,7 @@ use Workflow\V2\StartOptions; use Workflow\V2\WorkflowStub; use Workflow\WorkflowOptions; +require_once __DIR__.'/lock-pressure-middleware.php'; require __DIR__.'/selected-run-signal-state.php'; @@ -70,7 +71,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 +85,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]; diff --git a/scripts/conformance/server-parity/sdk-update-direction-probe.php b/scripts/conformance/server-parity/sdk-update-direction-probe.php index 9fe3f7bb3..e8ce522c0 100644 --- a/scripts/conformance/server-parity/sdk-update-direction-probe.php +++ b/scripts/conformance/server-parity/sdk-update-direction-probe.php @@ -4,6 +4,7 @@ use Composer\InstalledVersions; use DurableWorkflow\Client; +require_once __DIR__.'/lock-pressure-middleware.php'; function sdkUpdateActorOpen(string $language, string $phase, array $fixture, array $options, string $runId, float $deadline): array { @@ -78,7 +79,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 +114,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(), diff --git a/scripts/conformance/server-parity/task-retry-probe.php b/scripts/conformance/server-parity/task-retry-probe.php index 42c81935b..fa05a5fd8 100644 --- a/scripts/conformance/server-parity/task-retry-probe.php +++ b/scripts/conformance/server-parity/task-retry-probe.php @@ -3,12 +3,13 @@ declare(strict_types=1); use DurableWorkflow\Client; +require_once __DIR__.'/lock-pressure-middleware.php'; function finishHttpTaskRetry(Client $admin, array $fixture, string $workflowId, string $queue, string $url): array { $definition = $fixture['workflow_task_retry']; $requests = []; - $client = new Client($url, namespace: 'default', token: getenv('DW_PARITY_TOKEN') ?: null, + $client = parityClient($url, namespace: 'default', token: getenv('DW_PARITY_TOKEN') ?: null, transport: namespaceTransport($requests)); $workerId = $workflowId.'-retry-worker'; $peerId = $workflowId.$definition['peer_suffix']; diff --git a/scripts/conformance/server-parity/waiting-history-probe.php b/scripts/conformance/server-parity/waiting-history-probe.php index 6d21e0026..26d4efd99 100644 --- a/scripts/conformance/server-parity/waiting-history-probe.php +++ b/scripts/conformance/server-parity/waiting-history-probe.php @@ -3,12 +3,13 @@ declare(strict_types=1); use DurableWorkflow\Client; +require_once __DIR__.'/lock-pressure-middleware.php'; function finishHttpWaitingHistory(Client $admin, array $fixture, string $workflowId, string $queue, string $url): array { $definition = $fixture['waiting_for_history']; $requests = []; - $client = new Client($url, namespace: 'default', token: getenv('DW_PARITY_TOKEN') ?: null, + $client = parityClient($url, namespace: 'default', token: getenv('DW_PARITY_TOKEN') ?: null, transport: namespaceTransport($requests)); $workerId = $workflowId.'-waiting-worker'; $peerId = $workflowId.$definition['peer_suffix']; diff --git a/tests/Unit/ServerParityLockPressureRetryTest.php b/tests/Unit/ServerParityLockPressureRetryTest.php index e29e3beb9..849e1b310 100644 --- a/tests/Unit/ServerParityLockPressureRetryTest.php +++ b/tests/Unit/ServerParityLockPressureRetryTest.php @@ -50,6 +50,30 @@ public function test_other_requests_statuses_and_bodies_are_never_retried(): voi } } + public function test_documented_operation_refusals_are_retried_and_polls_keep_their_rules(): void + { + $deferred = ['reason' => 'backend_lock_pressure', 'outcome' => 'deferred', 'retryable' => true, 'retry_after_seconds' => 1]; + $complete = 'http://parity-server:8080/api/worker/workflow-tasks/task/complete'; + $start = 'http://parity-server:8080/api/workflows'; + self::assertSame(1.0, \lockPressureOperationRetryDelay('POST', $complete, 503, $deferred, 0, 0.0)); + self::assertSame(1.0, \lockPressureOperationRetryDelay('POST', $start, 503, + ['reason' => 'backend_lock_pressure', 'retryable' => true, 'retry_after_seconds' => 1], 0, 0.0)); + self::assertSame(1.0, \lockPressureOperationRetryDelay('POST', self::POLL, 503, $this->refusal(), 0, 0.0)); + foreach ([ + [$complete, 503, array_replace($deferred, ['retryable' => false])], + [$complete, 503, array_diff_key($deferred, ['retryable' => true])], + [$complete, 503, array_replace($deferred, ['reason' => 'backend_unavailable'])], + [$complete, 409, $deferred], + [$complete, 503, null], + // A poll refusal still needs task null, even through this entry point. + [self::POLL, 503, $deferred], + ] as $index => [$uri, $status, $response]) { + self::assertNull(\lockPressureOperationRetryDelay('POST', $uri, $status, $response, 0, 0.0), (string) $index); + } + self::assertNull(\lockPressureOperationRetryDelay('POST', $complete, 503, $deferred, 15, 0.0)); + self::assertNull(\lockPressureOperationRetryDelay('POST', $complete, 503, $deferred, 2, 29.5)); + } + public function test_retries_stop_at_the_attempt_and_wait_budgets(): void { self::assertNull(\lockPressurePollRetryDelay('POST', self::POLL, 503, $this->refusal(), 15, 0.0)); From d867e0f43955e591874b5ea9da28964f7578781b Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 10:39:48 +0000 Subject: [PATCH 2/5] Print failed Rust tests in the failing step's own log The parity jobs write Rust test output to evidence files, so a failed cargo test left only its exit code in the step log. An ERR trap now prints the panics and the failed test names from those files. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/server-parity.yml | 3 +++ .../conformance/server-parity/rust-test-failures.sh | 13 +++++++++++++ 2 files changed, 16 insertions(+) create mode 100755 scripts/conformance/server-parity/rust-test-failures.sh diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index c4bb53b6c..38d3bae9d 100644 --- a/.github/workflows/server-parity.yml +++ b/.github/workflows/server-parity.yml @@ -377,6 +377,7 @@ jobs: - name: Bootstrap independent published PHP and native MySQL databases run: | set -euo pipefail + trap 'bash scripts/conformance/server-parity/rust-test-failures.sh' ERR mkdir -p cargo-home rust-target parity-evidence mysql-state chmod 0777 parity-evidence docker network create "$RESOURCE_SCOPE" @@ -856,6 +857,7 @@ jobs: POSTGRES_IMAGE: ${{ matrix.image }} run: | set -euo pipefail + trap 'bash scripts/conformance/server-parity/rust-test-failures.sh' ERR mkdir -p cargo-home rust-target parity-evidence pg-state chmod 0777 parity-evidence docker network create "$RESOURCE_SCOPE" @@ -1364,6 +1366,7 @@ jobs: - name: Execute official Rust codec and cross-decode with published PHP run: | set -euo pipefail + trap 'bash scripts/conformance/server-parity/rust-test-failures.sh' ERR mkdir -p cargo-home rust-target parity-evidence # Compiling the library tests peaks near 2.5 GB; a 2 GB limit OOM-killed rustc (exit 101). docker run --rm --user="$(id -u):$(id -g)" --cpus=2 --memory=4g --memory-swap=4g \ diff --git a/scripts/conformance/server-parity/rust-test-failures.sh b/scripts/conformance/server-parity/rust-test-failures.sh new file mode 100755 index 000000000..6e83f9e9c --- /dev/null +++ b/scripts/conformance/server-parity/rust-test-failures.sh @@ -0,0 +1,13 @@ +#!/usr/bin/env bash +# Print the failures in this job's Rust test output, so a failed test step is +# diagnosable from its own log. Reads parity-evidence; changes nothing. +set -uo pipefail +for output in parity-evidence/*tests*.txt; do + [[ -f $output ]] && grep -q '^test result: FAILED' "$output" || continue + echo "::group::Rust test failures in ${output#parity-evidence/}" + grep -E -A3 'panicked at' "$output" | cut -c1-600 | head -n 120 + # The closing summary names every failed test. + sed -n '/^failures:$/,$p' "$output" | grep -E '^failures:$|^ [A-Za-z0-9_:]+$|^test result:' | tail -n 60 + echo "::endgroup::" +done +exit 0 From d94b1180e24576f33dae03c64bb9adfed11ffaa8 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 11:03:51 +0000 Subject: [PATCH 3/5] Call the Rust test failure summary through a named trap function The Docker workflow isolation policy only accepts repository helpers in statically known positions, as the container log tails already use. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/server-parity.yml | 18 +++++++++++++++--- 1 file changed, 15 insertions(+), 3 deletions(-) diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index 38d3bae9d..31788a264 100644 --- a/.github/workflows/server-parity.yml +++ b/.github/workflows/server-parity.yml @@ -377,7 +377,11 @@ jobs: - name: Bootstrap independent published PHP and native MySQL databases run: | set -euo pipefail - trap 'bash scripts/conformance/server-parity/rust-test-failures.sh' ERR + # Failed Rust tests are printed from their evidence files into this step's log. + rust_test_failures() { + bash scripts/conformance/server-parity/rust-test-failures.sh + } + trap rust_test_failures ERR mkdir -p cargo-home rust-target parity-evidence mysql-state chmod 0777 parity-evidence docker network create "$RESOURCE_SCOPE" @@ -857,7 +861,11 @@ jobs: POSTGRES_IMAGE: ${{ matrix.image }} run: | set -euo pipefail - trap 'bash scripts/conformance/server-parity/rust-test-failures.sh' ERR + # Failed Rust tests are printed from their evidence files into this step's log. + rust_test_failures() { + bash scripts/conformance/server-parity/rust-test-failures.sh + } + trap rust_test_failures ERR mkdir -p cargo-home rust-target parity-evidence pg-state chmod 0777 parity-evidence docker network create "$RESOURCE_SCOPE" @@ -1366,7 +1374,11 @@ jobs: - name: Execute official Rust codec and cross-decode with published PHP run: | set -euo pipefail - trap 'bash scripts/conformance/server-parity/rust-test-failures.sh' ERR + # Failed Rust tests are printed from their evidence files into this step's log. + rust_test_failures() { + bash scripts/conformance/server-parity/rust-test-failures.sh + } + trap rust_test_failures ERR mkdir -p cargo-home rust-target parity-evidence # Compiling the library tests peaks near 2.5 GB; a 2 GB limit OOM-killed rustc (exit 101). docker run --rm --user="$(id -u):$(id -g)" --cpus=2 --memory=4g --memory-swap=4g \ From fe2b26e6f64e2b4b5ad7d4e14f6ee4b0a9dceaf2 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 11:45:56 +0000 Subject: [PATCH 4/5] Retry lock-pressure refusals inside the probes' recorders Guzzle wraps a handler stack in reverse, so HandlerStack::unshift put the retry outermost and each recorder logged the refused attempt as well as the retried reply. The patch deployment contract then rejected a published PHP activity-poll refusal it did not expect. The retry is now pushed after each recorder, which makes it innermost, so recorders see only the final reply, as the middleware intended. Co-Authored-By: Claude Opus 5.5 --- .../conformance/server-parity/cooperative-probe.php | 2 +- .../server-parity/lock-pressure-middleware.php | 10 ++++++---- scripts/conformance/server-parity/namespace-probe.php | 2 +- .../server-parity/patch-deployment-probe.php | 2 +- scripts/conformance/server-parity/probe.php | 2 +- 5 files changed, 10 insertions(+), 8 deletions(-) diff --git a/scripts/conformance/server-parity/cooperative-probe.php b/scripts/conformance/server-parity/cooperative-probe.php index b856b308c..3a52e3ee4 100644 --- a/scripts/conformance/server-parity/cooperative-probe.php +++ b/scripts/conformance/server-parity/cooperative-probe.php @@ -26,7 +26,6 @@ function cooperativeObservedTransport(array &$traffic): Psr18Transport { $stack = HandlerStack::create(); - $stack->unshift(LockPressurePollRetries::middleware(), 'lock_pressure_poll_retry'); $stack->push(static function (callable $handler) use (&$traffic): callable { return static function (RequestInterface $request, array $options) use ($handler, &$traffic) { return $handler($request, $options)->then(static function (ResponseInterface $response) use ($request, &$traffic): ResponseInterface { @@ -44,6 +43,7 @@ function cooperativeObservedTransport(array &$traffic): Psr18Transport }); }; }); + $stack->push(LockPressurePollRetries::middleware(), 'lock_pressure_poll_retry'); return new Psr18Transport(client: new GuzzleClient(['http_errors' => false, 'handler' => $stack])); } diff --git a/scripts/conformance/server-parity/lock-pressure-middleware.php b/scripts/conformance/server-parity/lock-pressure-middleware.php index 11cf1d5ef..1ced25c8d 100644 --- a/scripts/conformance/server-parity/lock-pressure-middleware.php +++ b/scripts/conformance/server-parity/lock-pressure-middleware.php @@ -19,9 +19,11 @@ * (see lockPressureOperationRetryDelay and lockPressurePollRetryDelay for the * exact conditions). This Guzzle middleware retries only that refusal, below the SDK transport, so * the transport keeps every capability the SDK checks (bounded requests, - * payload transfer). It stops within the request's own timeout. Push it - * innermost (HandlerStack::unshift) so recording middleware sees only the final - * reply and observations stay comparable across targets. Retries are reported + * payload transfer). It stops within the request's own timeout. Push it after + * any recording middleware (HandlerStack::push), which makes it innermost: + * Guzzle wraps the stack in reverse, so unshift would put it outermost and let + * recorders see each refused attempt. Recorders then see only the final reply + * and observations stay comparable across targets. Retries are reported * once on stderr when the process exits. */ final class LockPressurePollRetries @@ -80,7 +82,7 @@ public static function middleware(): callable public static function transport(): Psr18Transport { $stack = HandlerStack::create(); - $stack->unshift(self::middleware(), 'lock_pressure_poll_retry'); + $stack->push(self::middleware(), 'lock_pressure_poll_retry'); return new Psr18Transport(client: new GuzzleClient(['http_errors' => false, 'handler' => $stack])); } diff --git a/scripts/conformance/server-parity/namespace-probe.php b/scripts/conformance/server-parity/namespace-probe.php index 4f756671d..f6d1e0fa1 100644 --- a/scripts/conformance/server-parity/namespace-probe.php +++ b/scripts/conformance/server-parity/namespace-probe.php @@ -46,7 +46,6 @@ function namespaceHttpRun(Client $client, string $workflowId, string $runId): ar function namespaceTransport(array &$receipts, ?Closure $beforeCompletion = null): Psr18Transport { $stack = HandlerStack::create(); - $stack->unshift(LockPressurePollRetries::middleware(), 'lock_pressure_poll_retry'); $stack->push(static function (callable $handler) use (&$receipts, $beforeCompletion): callable { return static function (RequestInterface $request, array $options) use ($handler, &$receipts, $beforeCompletion) { $path = $request->getUri()->getPath(); @@ -68,6 +67,7 @@ function namespaceTransport(array &$receipts, ?Closure $beforeCompletion = null) }); }; }); + $stack->push(LockPressurePollRetries::middleware(), 'lock_pressure_poll_retry'); return new Psr18Transport(client: new NamespaceHttp(['http_errors' => false, 'handler' => $stack, 'connect_timeout' => 1, 'timeout' => 5])); diff --git a/scripts/conformance/server-parity/patch-deployment-probe.php b/scripts/conformance/server-parity/patch-deployment-probe.php index 204b073c4..030b34773 100644 --- a/scripts/conformance/server-parity/patch-deployment-probe.php +++ b/scripts/conformance/server-parity/patch-deployment-probe.php @@ -101,7 +101,6 @@ function patchHttpPhase(array $fixture, array $options): array $worker = $handle = null; $requests = $diagnostics = []; $stack = HandlerStack::create(); - $stack->unshift(LockPressurePollRetries::middleware(), 'lock_pressure_poll_retry'); $stack->push(static function (callable $handler) use (&$worker, &$requests, $phase, $fixture): callable { return static function (RequestInterface $request, array $transportOptions) use ($handler, &$worker, &$requests, $phase, $fixture) { return $handler($request, $transportOptions)->then(static function (ResponseInterface $response) use ($request, &$worker, &$requests, $phase, $fixture): ResponseInterface { @@ -128,6 +127,7 @@ function patchHttpPhase(array $fixture, array $options): array }); }; }); + $stack->push(LockPressurePollRetries::middleware(), 'lock_pressure_poll_retry'); $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') { diff --git a/scripts/conformance/server-parity/probe.php b/scripts/conformance/server-parity/probe.php index 9c078e331..060d26745 100644 --- a/scripts/conformance/server-parity/probe.php +++ b/scripts/conformance/server-parity/probe.php @@ -71,7 +71,6 @@ function observedChildTransport(array &$polls, array &$completions, array &$acti // Observe real published transport I/O. No response or request is changed, // and authentication headers are never retained. $stack = HandlerStack::create(); - $stack->unshift(LockPressurePollRetries::middleware(), 'lock_pressure_poll_retry'); $stack->push(static function (callable $handler) use (&$polls, &$completions, &$activityPolls, &$activityOutcomes, $afterCompletion, &$scheduleReceipts): callable { return static function (RequestInterface $request, array $options) use ($handler, &$polls, &$completions, &$activityPolls, &$activityOutcomes, $afterCompletion, &$scheduleReceipts) { return $handler($request, $options)->then(static function (ResponseInterface $response) use ($request, &$polls, &$completions, &$activityPolls, &$activityOutcomes, $afterCompletion, &$scheduleReceipts): ResponseInterface { @@ -108,6 +107,7 @@ function observedChildTransport(array &$polls, array &$completions, array &$acti }); }; }); + $stack->push(LockPressurePollRetries::middleware(), 'lock_pressure_poll_retry'); return new Psr18Transport(client: new GuzzleClient(['http_errors' => false, 'handler' => $stack])); } From a25fa493542f2441e0f96fc6ffdd45d379394fd0 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 13:33:36 +0000 Subject: [PATCH 5/5] Keep the activity deadline probe's own poll retry accounting This probe retries lock-pressure poll refusals itself and records each retry, while a recording proxy outside the client logs every refused poll. Retrying below the SDK hid those retries from the probe, so the contract found a refused poll without its recorded retry. The probe keeps its plain client. Co-Authored-By: Claude Opus 5.5 --- scripts/conformance/server-parity/activity-deadline-probe.php | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/scripts/conformance/server-parity/activity-deadline-probe.php b/scripts/conformance/server-parity/activity-deadline-probe.php index 4728f6264..b4bd61dfe 100644 --- a/scripts/conformance/server-parity/activity-deadline-probe.php +++ b/scripts/conformance/server-parity/activity-deadline-probe.php @@ -3,7 +3,6 @@ declare(strict_types=1); use DurableWorkflow\Client; -require_once __DIR__.'/lock-pressure-middleware.php'; require_once __DIR__.'/namespace-probe.php'; require_once __DIR__.'/activity-deadline-poll.php'; @@ -35,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 = parityClient($url, namespace:'default', token:getenv('DW_PARITY_TOKEN') ?: null); + $client = new Client($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');