Retry the PHP reference's lock-pressure refusals in parity probes

.github/workflows/server-parity.yml

@@ -377,6 +377,11 @@ jobs:
       - name: Bootstrap independent published PHP and native MySQL databases
         run: |
           set -euo pipefail
+          # 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"
@@ -856,6 +861,11 @@ jobs:
           POSTGRES_IMAGE: ${{ matrix.image }}
         run: |
           set -euo pipefail
+          # 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"
@@ -1364,6 +1374,11 @@ jobs:
       - name: Execute official Rust codec and cross-decode with published PHP
         run: |
           set -euo pipefail
+          # 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 \

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 {

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),

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;

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

@@ -21,6 +21,7 @@
 use Workflow\V2\WorkflowStub;
 use Workflow\V2\StartOptions;
 use Workflow\WorkflowOptions;
+require_once __DIR__.'/lock-pressure-middleware.php';
 
 function cooperativeObservedTransport(array &$traffic): Psr18Transport
 {
@@ -42,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]));
 }
@@ -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,

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);

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<Duration> {
+    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<T, F, Fut>(what: &str, mut call: F) -> Result<T>
+where
+    F: FnMut() -> Fut,
+    Fut: Future<Output = Result<T>>,
+{
+    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)));
+    }
+}

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(

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!(

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

@@ -0,0 +1,97 @@
+<?php
+
+declare(strict_types=1);
+
+use DurableWorkflow\Client;
+use DurableWorkflow\Transport\Psr18Transport;
+use GuzzleHttp\Client as GuzzleClient;
+use GuzzleHttp\HandlerStack;
+use GuzzleHttp\Promise\Create;
+use GuzzleHttp\Psr7\Utils;
+use Psr\Http\Message\RequestInterface;
+use Psr\Http\Message\ResponseInterface;
+
+require_once __DIR__.'/lock-pressure-retry.php';
+
+/**
+ * The published PHP SQLite reference shares its file with its maintenance loop
+ * and may refuse a request with the documented 503 backend_lock_pressure reply
+ * (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 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
+{
+    /** @var list<array{path: string, retry_after_seconds: float}> */
+    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->push(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);
+}