Retry questionBefore adding a retry, identify the durable request ID, the side effect's idempotency boundary, and what evidence proves the prior attempt did not already succeed.
+
+## Bound failure diagnostics
+
+Failure reporting keeps diagnostic content within finite defaults:
+
+| Content | Default budget |
+| --- | --- |
+| UTF-8 message, including its marker | 62 KiB |
+| Stack trace, including diagnostic markers | 128 KiB |
+| Serialized details and diagnostics together | 64 KiB |
+| Complete encoded failure request, including JSON escaping and identity | 1 MiB |
+
+The message budget reserves 2 KiB below the original 64 KiB ceiling for
+published runtime summaries. It does not repair the separate runtime summary
+byte-bound defect tracked in shared issue 225.
+
+Oversized content produces a visible `[truncated failure report; ...]` message
+marker and bounded diagnostics with original sizes and omitted fields. The SDK
+retains a complete valid Avro details envelope or explicitly omits it. It never
+slices Avro or base64 bytes. Small reports keep their message and details.
+
+The original failure type, retry decision, authored retry delays and claimed
+attempt identity remain unchanged. Diagnostic limits are separate from business
+payload and worker request timeout budgets. Cooperative callbacks also bound
+failure diagnostics before sending them to their owning worker.
+
+For a transport with a smaller request budget, narrow the client limit:
+
+```php
+$client = $client->withFailureReportLimit(128 * 1024);
+```
+
+An injected transport can expose a lower known body budget through
+`RequestSizeLimitTransport`. The lower limit applies after markers and JSON
+escaping are included. If required identity or fixed request content cannot fit,
+`FailureReportRefused` reports `failure_report_refused` locally. Treat this as a
+reporting refusal requiring configuration or identity repair, rather than a new
+application failure or permission to publish under a different attempt.
+
+Successful local records keep their normal business-payload budget. When a
+completion includes both failure diagnostics and a large business result, the
+existing runtime upload protocol preserves the complete Avro payload while
+keeping the final failure request bounded. If fixed content cannot fit and the
+namespace cannot store that payload, reporting refuses explicitly. No partial
+business payload is published.
+
+Local callback failures normalize before workflow code branches on them. The
+first execution, persisted local record and fresh replay use the same bounded
+message, type and retry decision.
+
+Activity failures use existing diagnostic fields. Workflow-task failures use
+the existing stack string for structured truncation information. The published
+Server qualification records task failures and grants a legitimate retry turn,
+but that runtime does not append a workflow-task failure event to workflow
+history or persist its stack diagnostics. That visibility work remains tracked
+in [shared issue 225](https://github.com/durable-workflow/.github/issues/225).
diff --git a/docs/quickstart-contract.json b/docs/quickstart-contract.json
index d98883b6..824e1d14 100644
--- a/docs/quickstart-contract.json
+++ b/docs/quickstart-contract.json
@@ -3,9 +3,9 @@
"schema_version": 2,
"package": {
"name": "durable-workflow/sdk",
- "published_version": "3.4.1",
- "composer_requirement": "3.4.1",
- "onboarding_requirement": "^3.4"
+ "published_version": "3.5.0",
+ "composer_requirement": "3.5.0",
+ "onboarding_requirement": "^3.5"
},
"runtime_targets": {
"server": {
diff --git a/scripts/ci/run-replay-regression-fixture.php b/scripts/ci/run-replay-regression-fixture.php
index ebb7c387..1663c4b0 100644
--- a/scripts/ci/run-replay-regression-fixture.php
+++ b/scripts/ci/run-replay-regression-fixture.php
@@ -377,6 +377,7 @@ public function run(WorkflowContext $context): array
final class ReplayRegressionConsumer
{
private const FIXTURE_SCHEMA = 'durable-workflow.replay-regression/v1';
+ private const BOUNDED_LOCAL_FAILURE = 'golden.bounded-local-failure';
public static function executeFile(string $path): void
{
@@ -554,26 +555,7 @@ private static function executeLayout(
);
}
$localActivityInvocations = 0;
- if (in_array($workflowType, ['golden.local-activity-terminal-failure', 'golden.local-activity-recovered'], true)) {
- $worker->registerActivity(
- $workflowType === 'golden.local-activity-recovered' ? 'golden.local-receipt' : 'golden.local-failure',
- static function (ActivityContext $_context) use (&$localActivityInvocations): never {
- ++$localActivityInvocations;
-
- throw new RuntimeException('Cold replay repeated the recorded local activity side effect.');
- },
- );
- }
- if ($workflowType === 'golden.activity-completion-backend-loss') {
- $worker->registerActivity(
- 'golden.backend-recovery',
- static function (ActivityContext $_context) use (&$localActivityInvocations): string {
- ++$localActivityInvocations;
-
- return 'activity-after-backend-recovery';
- },
- );
- }
+ self::registerLocalActivities($worker, $workflowType, $localActivityInvocations);
foreach ($tasks as $_task) {
$worker->tick(0);
gc_collect_cycles();
@@ -581,7 +563,7 @@ static function (ActivityContext $_context) use (&$localActivityInvocations): st
if ($workflowType === 'golden.workflow-lease-recovery') {
$worker->tick(0);
}
- if ($localActivityInvocations !== ($workflowType === 'golden.activity-completion-backend-loss' ? 1 : 0)) {
+ if ($localActivityInvocations !== (in_array($workflowType, ['golden.activity-completion-backend-loss', self::BOUNDED_LOCAL_FAILURE], true) ? 1 : 0)) {
throw new RuntimeException("{$identity} repeated a recorded local activity during cold replay.");
}
@@ -640,6 +622,40 @@ private static function assertExpected(string $identity, array $fixture, array $
self::assertMatches($expected, $observed, "{$identity}.expected");
}
+ private static function registerLocalActivities(Worker $worker, string $workflowType, int &$localActivityInvocations): void
+ {
+ if (in_array($workflowType, ['golden.local-activity-terminal-failure', 'golden.local-activity-recovered'], true)) {
+ $worker->registerActivity(
+ $workflowType === 'golden.local-activity-recovered' ? 'golden.local-receipt' : 'golden.local-failure',
+ static function (ActivityContext $_context) use (&$localActivityInvocations): never {
+ ++$localActivityInvocations;
+
+ throw new RuntimeException('Cold replay repeated the recorded local activity side effect.');
+ },
+ );
+ }
+ if ($workflowType === self::BOUNDED_LOCAL_FAILURE) {
+ $worker->registerActivity(self::BOUNDED_LOCAL_FAILURE,
+ static function (ActivityContext $context) use (&$localActivityInvocations): never {
+ if ($context->activityType !== self::BOUNDED_LOCAL_FAILURE) {
+ throw new UnexpectedValueException('Unexpected bounded-local activity context.');
+ }
+ ++$localActivityInvocations;
+ throw new RuntimeException(str_repeat('雪', 1024 * 1024));
+ });
+ }
+ if ($workflowType === 'golden.activity-completion-backend-loss') {
+ $worker->registerActivity(
+ 'golden.backend-recovery',
+ static function (ActivityContext $_context) use (&$localActivityInvocations): string {
+ ++$localActivityInvocations;
+
+ return 'activity-after-backend-recovery';
+ },
+ );
+ }
+ }
+
/** @return callable(WorkflowContext, mixed ...$input): mixed */
private static function workflow(string $workflowType): callable
{
@@ -872,6 +888,16 @@ private static function workflow(string $workflowType): callable
return ['key' => $selected->key];
},
+ self::BOUNDED_LOCAL_FAILURE => static function (WorkflowContext $context): array {
+ try {
+ $context->localActivity('golden.bounded-local-failure', [], ['retry_policy' => ['max_attempts' => 1]]);
+ return ['unexpected_success' => true];
+ } catch (ActivityFailed $failure) {
+ return ['marked' => str_contains($failure->getMessage(), '[truncated failure report;'),
+ 'message_bytes' => strlen($failure->getMessage()), 'failure_type' => $failure->failureType,
+ 'non_retryable' => $failure->nonRetryable];
+ }
+ },
'golden.local-activity-recovered' => static fn (WorkflowContext $context): mixed =>
$context->localActivity('golden.local-receipt'),
'golden.local-activity-terminal-failure' => static function (WorkflowContext $context): never {
diff --git a/scripts/qualify-failure-report-delivery.php b/scripts/qualify-failure-report-delivery.php
new file mode 100644
index 00000000..b492a2d9
--- /dev/null
+++ b/scripts/qualify-failure-report-delivery.php
@@ -0,0 +1,324 @@
+describeWorkflow($handle->workflowId, $handle->selectedRunId);
+ if ($execution->isTerminal === true || in_array($execution->status,
+ ['completed', 'failed', 'cancelled', 'terminated', 'timed_out'], true)) {
+ return $handle->resultOfSelectedRun(1, 0.05);
+ }
+ $maintain();
+ $worker->tick(0);
+ $activities?->tick(0);
+ usleep(10000);
+ } while (microtime(true) < $deadline);
+ throw new FailureDeliveryQualificationException('The selected run did not finish within its fixture budget.');
+}
+
+$http = new class implements Psr\Http\Client\ClientInterface {
+ public array $reports = [];
+ public array $claims = [];
+ private GuzzleHttp\Client $client;
+ public function __construct() { $this->client = new GuzzleHttp\Client(['http_errors' => false, 'timeout' => 15]); }
+ public function sendRequest(Psr\Http\Message\RequestInterface $request): Psr\Http\Message\ResponseInterface
+ {
+ $response = $this->client->sendRequest($request);
+ $this->observe($request, $response);
+ return $response;
+ }
+ public function observe(Psr\Http\Message\RequestInterface $request, Psr\Http\Message\ResponseInterface $response): void
+ {
+ $wire = (string) $request->getBody();
+ $data = $wire === '' ? [] : json_decode($wire, true, flags: JSON_THROW_ON_ERROR);
+ if ($response->getStatusCode() >= 400) {
+ fwrite(STDERR, json_encode(['path' => $request->getUri()->getPath(), 'status' => $response->getStatusCode(),
+ 'response' => array_diff_key(json_decode((string) $response->getBody(), true), ['server_capabilities' => true])], JSON_THROW_ON_ERROR)."\n");
+ }
+ $path = $request->getUri()->getPath();
+ if (str_ends_with($path, '/fail') || isset($data['commands'][0]['exception_type'])) {
+ $failure = $data['failure'] ?? $data['commands'][0];
+ $this->reports[] = ['path' => $path, 'encoded_bytes' => strlen($wire), 'http_status' => $response->getStatusCode(),
+ 'lease_owner' => $data['lease_owner'] ?? null, 'activity_attempt_id' => $data['activity_attempt_id'] ?? null,
+ 'workflow_task_attempt' => $data['workflow_task_attempt'] ?? null,
+ 'type' => isset($data['failure']) ? ($failure['type'] ?? null) : ($failure['exception_type'] ?? null),
+ 'message_bytes' => strlen($failure['message']),
+ 'marked' => str_contains($failure['message'], FAILURE_DELIVERY_MARKER),
+ 'stack_bytes' => strlen($failure['stack_trace'] ?? $failure['exception']['stack_trace'] ?? ''),
+ 'stack_includes_qualifier' => str_contains($failure['stack_trace'] ?? $failure['exception']['stack_trace'] ?? '', 'qualify-failure-report-delivery.php'),
+ 'diagnostics' => $failure['diagnostics'] ?? $failure['exception']['diagnostics'] ?? null,
+ 'message_sha256' => hash('sha256', $failure['message']), 'non_retryable' => $failure['non_retryable'] ?? null];
+ }
+ if (str_ends_with($path, '/poll')) {
+ $reply = json_decode((string) $response->getBody(), true, flags: JSON_THROW_ON_ERROR);
+ $task = $reply['task'] ?? null;
+ if (is_array($task)) {
+ $this->claims[] = array_intersect_key($task, array_flip([
+ 'task_id', 'lease_owner', 'workflow_task_attempt', 'activity_attempt_id', 'attempt_id', 'workflow_id', 'run_id',
+ ]));
+ }
+ }
+ }
+};
+$namespace = 'php-failure-'.bin2hex(random_bytes(6));
+$admin = new Client($url, controlToken: 'claimed-task-admin');
+$admin->createNamespace($namespace);
+$control = new Client($url, namespace: $namespace, controlToken: 'claimed-task-client');
+$transport = new DurableWorkflow\Transport\Psr18Transport(client: $http);
+$client = new Client($url, namespace: $namespace, transport: $transport, workerToken: 'claimed-task-worker');
+$queue = $namespace;
+$workerId = $namespace.'-workflow';
+$activityWorkerId = $namespace.'-activity';
+$worker = new Worker($client, $queue, workerId: $workerId, stickyCacheCapacity: 1);
+$workflowHandler = static function (WorkflowContext $context, string $kind): array {
+ if ($kind === 'terminal') {
+ throw new FailureDeliveryQualificationException(str_repeat('t', 3 * 1024 * 1024));
+ }
+ if ($kind === 'retry') {
+ return ['activity_value' => $context->activity('qualification.failure.activity', [],
+ ['retry_policy' => ['max_attempts' => 2, 'backoff_seconds' => [1]]])];
+ }
+ if ($kind === 'workflow-task') {
+ return ['fresh_turn' => true];
+ }
+ try {
+ if ($kind === 'local-failure') {
+ $context->localActivity('qualification.failure.local', [], ['retry_policy' => ['max_attempts' => 1]]);
+ } else {
+ $context->activity($kind === 'cooperative-worker' ? 'qualification.failure.managed' : 'qualification.failure.activity',
+ [], ['retry_policy' => ['max_attempts' => 1]]);
+ }
+ throw new FailureDeliveryQualificationException('The activity unexpectedly succeeded.');
+ } catch (ActivityFailed $failure) {
+ return ['type' => $failure->failureType, 'marked' => str_contains($failure->getMessage(), FAILURE_DELIVERY_MARKER),
+ 'message_bytes' => strlen($failure->getMessage()), 'message_sha256' => hash('sha256', $failure->getMessage()),
+ 'non_retryable' => $failure->nonRetryable];
+ }
+};
+$worker->registerWorkflow('qualification.failure', $workflowHandler);
+$worker->registerActivity('qualification.failure.local', static function (DurableWorkflow\Worker\ActivityContext $_context): never {
+ requireDelivery($_context->activityType === 'qualification.failure.local', 'Wrong local callback context.');
+ throw new FailureDeliveryQualificationException(str_repeat('雪', 1024 * 1024));
+});
+$client->registerWorker($workerId, $queue, ['qualification.failure'], [], ['sticky_execution', 'typed_search_attributes', 'local_activities'],
+ workflowCommandContracts: $worker->contracts()['workflow_commands']);
+$client->registerWorker($activityWorkerId, $queue, [], ['qualification.failure.activity']);
+$registeredWorkers = [$workerId => $client, $activityWorkerId => $client];
+$lastHeartbeats = [];
+$maintenance = [];
+$maintain = static function () use (&$registeredWorkers, &$lastHeartbeats, &$maintenance): void {
+ foreach ($registeredWorkers as $identity => $connection) {
+ $now = hrtime(true) / 1e9;
+ if ($now - ($lastHeartbeats[$identity] ?? 0) < 5) {
+ continue;
+ }
+ // tick() deliberately leaves registration lifecycle to its caller.
+ $reply = $connection->heartbeatWorker($identity);
+ $lastHeartbeats[$identity] = $now;
+ $maintenance[] = ['worker_id' => $identity, 'registration_id' => $reply['registration_id'] ?? null,
+ 'registration_incarnation' => $reply['registration_incarnation'] ?? null];
+ }
+};
+$result = ['sdk_version' => Composer\InstalledVersions::getPrettyVersion('durable-workflow/sdk'),
+ 'sdk_reference' => Composer\InstalledVersions::getReference('durable-workflow/sdk'),
+ 'server_version' => $admin->clusterInfo()->version, 'namespace' => $namespace, 'scenarios' => []];
+register_shutdown_function(static function () use (&$result, $http, &$maintenance): void {
+ if (!isset($result['pass'])) {
+ $result['reports'] = $http->reports;
+ $result['claims'] = $http->claims;
+ $result['maintenance'] = $maintenance;
+ $result['incomplete'] = true;
+ $result['last_error'] = error_get_last();
+ echo json_encode($result, JSON_THROW_ON_ERROR | JSON_PRETTY_PRINT | JSON_UNESCAPED_UNICODE)."\n";
+ }
+});
+$large = str_repeat('m', 3 * 1024 * 1024);
+foreach (['message', 'stack', 'combined', 'escaped', 'details', 'lower-limit', 'lower-details', 'small'] as $kind) {
+ $handle = $control->startWorkflow('qualification.failure', $namespace.'-'.$kind, $queue, [$kind], runTimeoutSeconds: 120);
+ $maintain();
+ $worker->tick(0);
+ $task = $client->pollActivityTask($activityWorkerId, $queue, 0);
+ requireDelivery(is_array($task), 'The activity was not leased for '.$kind);
+ $message = in_array($kind, ['message', 'combined', 'lower-limit'], true) ? $large : 'original message';
+ $stack = in_array($kind, ['stack', 'combined', 'lower-limit'], true) ? $large : null;
+ if ($kind === 'escaped') {
+ $message = $stack = str_repeat("\0\"\\雪🦀", 300000);
+ }
+ $details = match ($kind) {
+ 'details' => ['nested' => ['cause' => $large]],
+ 'lower-details' => ['text' => str_repeat('d', 45000)],
+ default => ['whole' => 1.0, 'zero' => -0.0],
+ };
+ $type = 'OriginalFailure.'.$kind;
+ $reporter = in_array($kind, ['lower-limit', 'lower-details'], true) ? $client->withFailureReportLimit(4096) : $client;
+ $reporter->failActivityTask($task['task_id'], $task['activity_attempt_id'] ?? $task['attempt_id'], $task['lease_owner'],
+ $message, $type, true, $details, $stack);
+ $maintain();
+ $worker->tick(0);
+ $output = finishFailureWorkflow($control, $worker, $handle, $maintain);
+ requireDelivery($output['type'] === $type && $output['marked'] === ($kind !== 'small'), 'Original workflow failure mismatch.');
+ requireDelivery($output['non_retryable'] === true, 'The non-retryable decision changed.');
+ $history = $control->workflowHistory($handle->workflowId, $handle->selectedRunId, 100);
+ $events = $history['events'] ?? $history['history_events'] ?? [];
+ $counts = array_count_values(array_column($events, 'event_type'));
+ requireDelivery(($counts['ActivityFailed'] ?? 0) === 1 && ($counts['WorkflowCompleted'] ?? 0) === 1, 'Failure history missing.');
+ $result['scenarios'][$kind] = ['workflow_id' => $handle->workflowId, 'run_id' => $handle->selectedRunId,
+ 'task_id' => $task['task_id'], 'attempt_id' => $task['activity_attempt_id'] ?? $task['attempt_id'],
+ 'lease_owner' => $task['lease_owner'], 'output' => $output, 'event_counts' => $counts];
+}
+$handle = $control->startWorkflow('qualification.failure', $namespace.'-local', $queue, ['local-failure'], runTimeoutSeconds: 120);
+$maintain();
+$worker->tick(0);
+$output = finishFailureWorkflow($control, $worker, $handle, $maintain);
+$history = $control->workflowHistory($handle->workflowId, $handle->selectedRunId, 100);
+$events = $history['events'] ?? $history['history_events'] ?? [];
+$failed = array_values(array_filter($events, static fn (array $event): bool => $event['event_type'] === 'ActivityFailed'));
+requireDelivery(count($failed) === 1, 'Local failure was not persisted once.');
+$message = $failed[0]['payload']['message'] ?? '';
+requireDelivery($output['message_sha256'] === hash('sha256', $message), 'Initial local branch saw a different failure from persisted history.');
+requireDelivery($output['type'] === FailureDeliveryQualificationException::class && $output['marked'] === true,
+ 'Original local failure type or marker changed.');
+$cold = (new DurableWorkflow\Worker\Replayer($client->payloadCodec()))->replay($workflowHandler, $events, ['local-failure'], $queue,
+ localActivityExecutor: static function (): never { throw new FailureDeliveryQualificationException('Cold replay repeated a local callback.'); });
+requireDelivery($cold->terminalFailure === null && count($cold->commands) === 1, 'Local cold replay did not complete normally.');
+requireDelivery($client->payloadCodec()->decodeEnvelope($cold->commands[0]['result']) === $output,
+ 'Cold workflow branch saw a different local failure.');
+$result['scenarios']['local-failure'] = ['workflow_id' => $handle->workflowId, 'run_id' => $handle->selectedRunId,
+ 'output' => $output, 'initial_equals_persisted_equals_cold' => true];
+$stack = GuzzleHttp\HandlerStack::create();
+$stack->push(static function (callable $next) use ($http): Closure {
+ return static function (Psr\Http\Message\RequestInterface $request, array $options) use ($next, $http) {
+ return $next($request, $options)->then(static function (Psr\Http\Message\ResponseInterface $response) use ($request, $http) {
+ $http->observe($request, $response);
+ return $response;
+ });
+ };
+});
+$managedId = $namespace.'-cooperative';
+$managedClient = new Client($url, namespace: $namespace, workerToken: 'claimed-task-worker',
+ transport: new DurableWorkflow\Transport\Psr18Transport(client: new GuzzleHttp\Client(['handler' => $stack, 'http_errors' => false])));
+$managedWorker = new Worker($managedClient, $queue, workerId: $managedId, enableCooperativeCancellation: true);
+$managedWorker->registerActivity('qualification.failure.managed', static function (DurableWorkflow\Worker\ActivityContext $_context): never {
+ requireDelivery($_context->activityType === 'qualification.failure.managed', 'Wrong isolated callback context.');
+ throw new FailureDeliveryQualificationException(str_repeat('m', 3 * 1024 * 1024));
+});
+$managedClient->registerWorker($managedId, $queue, [], ['qualification.failure.managed'], ['cooperative_cancellation']);
+$registeredWorkers[$managedId] = $managedClient;
+$handle = $control->startWorkflow('qualification.failure', $namespace.'-cooperative-worker', $queue, ['cooperative-worker'], runTimeoutSeconds: 120);
+$maintain();
+$worker->tick(0);
+$managedWorker->tick(0);
+$maintain();
+$worker->tick(0);
+$output = finishFailureWorkflow($control, $worker, $handle, $maintain, $managedWorker);
+requireDelivery($output['type'] === FailureDeliveryQualificationException::class && $output['marked'] === true,
+ 'The original workflow did not receive its isolated callback failure.');
+requireDelivery($output['non_retryable'] === false, 'The callback retry classification changed.');
+$history = $control->workflowHistory($handle->workflowId, $handle->selectedRunId, 100);
+$events = $history['events'] ?? $history['history_events'] ?? [];
+$failed = array_values(array_filter($events, static fn (array $event): bool => $event['event_type'] === 'ActivityFailed'));
+$failurePayload = $failed[0]['payload'] ?? [];
+requireDelivery(($failurePayload['exception']['diagnostics']['truncation']['fields']['message']['original_bytes'] ?? null) === 3145728,
+ 'Original callback size diagnostics missing from history.');
+$report = $http->reports[array_key_last($http->reports)];
+requireDelivery($report['stack_bytes'] > 0 && $report['stack_includes_qualifier'] === true,
+ 'The SDK did not transmit the original isolated callback trace.');
+$result['scenarios']['cooperative-worker'] = ['workflow_id' => $handle->workflowId, 'run_id' => $handle->selectedRunId,
+ 'output' => $output, 'original_size_in_history' => true, 'transmitted_original_stack_bytes' => $report['stack_bytes'],
+ 'history_stack_present' => isset($failurePayload['exception']['stack_trace']),
+ 'history_stack_visibility' => 'Published runtime omits the transmitted stack from public history; shared225 runtime follow-through remains.'];
+$handle = $control->startWorkflow('qualification.failure', $namespace.'-retry', $queue, ['retry'], runTimeoutSeconds: 120);
+$maintain();
+$worker->tick(0);
+$original = $client->pollActivityTask($activityWorkerId, $queue, 0);
+requireDelivery(is_array($original), 'Retry control original attempt was not leased.');
+$retryStarted = hrtime(true);
+$client->failActivityTask($original['task_id'], $original['activity_attempt_id'] ?? $original['attempt_id'],
+ $original['lease_owner'], $large, 'OriginalRetryableFailure', false);
+$deadline = microtime(true) + 15;
+do {
+ $retry = $client->pollActivityTask($activityWorkerId, $queue, 0);
+ if ($retry !== null) { break; }
+ usleep(50000);
+} while (microtime(true) < $deadline);
+requireDelivery(is_array($retry), 'Authored retry was not made available.');
+$originalAttempt = $original['activity_attempt_id'] ?? $original['attempt_id'];
+$retryAttempt = $retry['activity_attempt_id'] ?? $retry['attempt_id'];
+requireDelivery($retryAttempt !== $originalAttempt, 'The failed attempt was reused as publication authority.');
+$client->completeActivityTask($retry['task_id'], $retryAttempt, $retry['lease_owner'], 42);
+$maintain();
+$worker->tick(0);
+$output = finishFailureWorkflow($control, $worker, $handle, $maintain);
+requireDelivery($output === ['activity_value' => 42], 'The original workflow did not receive the legitimate retry result.');
+$history = $control->workflowHistory($handle->workflowId, $handle->selectedRunId, 100);
+$events = $history['events'] ?? $history['history_events'] ?? [];
+$scheduled = array_values(array_filter($events, static fn (array $event): bool => $event['event_type'] === 'ActivityScheduled'));
+$result['scenarios']['retry'] = ['workflow_id' => $handle->workflowId, 'run_id' => $handle->selectedRunId,
+ 'original_task_id' => $original['task_id'], 'original_attempt_id' => $originalAttempt,
+ 'retry_task_id' => $retry['task_id'], 'retry_attempt_id' => $retryAttempt,
+ 'retry_elapsed_seconds' => (hrtime(true) - $retryStarted) / 1e9,
+ 'scheduled_payload' => $scheduled[0]['payload'] ?? null, 'output' => $output];
+$handle = $control->startWorkflow('qualification.failure', $namespace.'-workflow-task', $queue, ['workflow-task'], runTimeoutSeconds: 120);
+$task = $client->pollWorkflowTask($workerId, $queue, 0);
+requireDelivery(is_array($task), 'Original workflow task not leased.');
+$ack = $client->failWorkflowTask($task['task_id'], $task['lease_owner'], $task['workflow_task_attempt'], $large,
+ 'OriginalWorkflowTaskFailure', stackTrace: $large);
+requireDelivery(($ack['recorded'] ?? null) === true, 'Workflow-task failure not recorded.');
+$maintain();
+$worker->tick(0);
+$output = finishFailureWorkflow($control, $worker, $handle, $maintain);
+requireDelivery($output === ['fresh_turn' => true], 'The workflow did not receive its fresh legitimate turn.');
+$history = $control->workflowHistory($handle->workflowId, $handle->selectedRunId, 100);
+$events = $history['events'] ?? $history['history_events'] ?? [];
+
+$failureEvents = array_values(array_filter($events, static fn (array $event): bool => $event['event_type'] === 'WorkflowTaskFailed'));
+// The published runtime stores task failure separately and emits no history event.
+requireDelivery(($ack['next_task_id'] ?? null) !== $task['task_id'], 'Failed claim identity was reused as retry authority.');
+$result['scenarios']['workflow-task'] = ['workflow_id' => $handle->workflowId, 'run_id' => $handle->selectedRunId,
+ 'original_task_id' => $task['task_id'], 'lease_owner' => $task['lease_owner'], 'attempt' => $task['workflow_task_attempt'],
+ 'ack' => $ack, 'history_failure_events' => count($failureEvents), 'fresh_turn_result' => $output,
+ 'history_visibility' => 'Published runtime emits no WorkflowTaskFailed history event; parent225 runtime follow-through remains.'];
+$handle = $control->startWorkflow('qualification.failure', $namespace.'-terminal', $queue, ['terminal'], runTimeoutSeconds: 120);
+$maintain();
+$worker->tick(0);
+try {
+ finishFailureWorkflow($control, $worker, $handle, $maintain);
+ throw new FailureDeliveryQualificationException('Terminal workflow unexpectedly succeeded.');
+} catch (WorkflowFailed $failure) {
+ requireDelivery($failure->failureType === FailureDeliveryQualificationException::class, 'Workflow failure type changed.');
+ requireDelivery(str_contains($failure->getMessage(), FAILURE_DELIVERY_MARKER), 'Terminal workflow marker missing.');
+ $result['scenarios']['terminal'] = ['workflow_id' => $handle->workflowId, 'run_id' => $handle->selectedRunId,
+ 'failure_type' => $failure->failureType, 'marked_message' => true, 'message_bytes' => strlen($failure->getMessage())];
+}
+foreach ($http->reports as $report) {
+ requireDelivery($report['encoded_bytes'] <= 1048576 && $report['http_status'] === 200, 'Encoded/reporting HTTP boundary failed.');
+}
+$result['reports'] = $http->reports;
+$result['claims'] = $http->claims;
+$result['maintenance'] = $maintenance;
+$result['pass'] = true;
+echo json_encode($result, JSON_THROW_ON_ERROR | JSON_PRETTY_PRINT | JSON_UNESCAPED_UNICODE)."\n";
diff --git a/scripts/qualify-failure-reports.php b/scripts/qualify-failure-reports.php
new file mode 100644
index 00000000..a83363eb
--- /dev/null
+++ b/scripts/qualify-failure-reports.php
@@ -0,0 +1,46 @@
+requests[] = $request;
+ return new GuzzleHttp\Psr7\Response(200, ['Content-Type' => 'application/json'], '{}');
+ }
+};
+$client = new DurableWorkflow\Client('https://recording.invalid',
+ transport: new DurableWorkflow\Transport\Psr18Transport(client: $http));
+$large = str_repeat('m', 3 * 1024 * 1024);
+$results = [];
+foreach (['activity', 'workflow-task'] as $kind) {
+ $http->requests = [];
+ if ($kind === 'activity') {
+ $client->failActivityTask('original-task', 'original-attempt', 'original-owner', $large, 'OriginalFailure', true);
+ } else {
+ $client->failWorkflowTask('original-task', 'original-owner', 3, $large, 'OriginalFailure');
+ }
+ $request = $http->requests[array_key_last($http->requests)];
+ $wire = (string) $request->getBody();
+ $body = json_decode($wire, true, flags: JSON_THROW_ON_ERROR);
+ $failure = $body['failure'];
+ $results[$kind] = ['original_message_bytes' => strlen($large), 'encoded_request_bytes' => strlen($wire),
+ 'message_bytes' => strlen($failure['message']), 'original_type' => $failure['type'] === 'OriginalFailure',
+ 'exact_identity' => $body['lease_owner'] === 'original-owner'
+ && ($kind === 'activity' ? $body['activity_attempt_id'] === 'original-attempt' : $body['workflow_task_attempt'] === 3),
+ 'marked_message' => str_contains($failure['message'], '[truncated failure report;'),
+ 'non_retryable' => $failure['non_retryable'] ?? null];
+}
+$pass = true;
+foreach ($results as $result) {
+ $pass = $pass && $result['original_type'] && $result['exact_identity']
+ && ($baseline ? $result['encoded_request_bytes'] > 2097152 && !$result['marked_message']
+ : $result['encoded_request_bytes'] <= 1048576 && $result['marked_message']);
+}
+echo json_encode(['baseline' => $baseline, 'sdk_version' => Composer\InstalledVersions::getPrettyVersion('durable-workflow/sdk'),
+ 'pass' => $pass, 'results' => $results], JSON_THROW_ON_ERROR | JSON_PRETTY_PRINT)."\n";
+exit($pass ? 0 : 1);
diff --git a/src/Client.php b/src/Client.php
index 26d58968..b086be97 100644
--- a/src/Client.php
+++ b/src/Client.php
@@ -9,6 +9,10 @@
use DurableWorkflow\Codec\AvroPayloadCodec;
use DurableWorkflow\Codec\PayloadCodec;
use DurableWorkflow\Exception\CodecException;
+use DurableWorkflow\Exception\FailureReportRefused;
+use DurableWorkflow\Failure\FailureReportLimiter;
+use DurableWorkflow\Failure\FailureReportLimits;
+use DurableWorkflow\Failure\LocalFailureReports;
use DurableWorkflow\Exception\QueryFailed;
use DurableWorkflow\Exception\ServerException;
use DurableWorkflow\Exception\SignalFailed;
@@ -74,6 +78,7 @@ final class Client implements WorkflowClientInterface
private readonly Transport $transport;
private readonly PayloadCodec $codec;
private readonly RuntimePayloadUploads $payloadUploads;
+ private FailureReportLimiter $failureReports;
private bool $boundedWorkerRequests = false;
private ActivityLeaseBinding $activityLeaseBinding;
private ?string $activityLeaseBootstrap = null;
@@ -134,9 +139,29 @@ public function __construct(
$this->transport = $transport ?? new Psr18Transport(requestTimeoutSeconds: $requestTimeoutSeconds);
$this->codec = $codec ?? new AvroPayloadCodec();
$this->payloadUploads = new RuntimePayloadUploads($this->transport, $this->baseUri);
+ $this->failureReports = new FailureReportLimiter(new FailureReportLimits(), $this->transport);
$this->activityLeaseBinding = new ActivityLeaseBinding();
}
+ /** Narrow the encoded failure-report request budget for an application transport. */
+ public function withFailureReportLimit(int $requestBytes): self
+ {
+ $copy = clone $this;
+ $copy->failureReports = new FailureReportLimiter(new FailureReportLimits($requestBytes), $this->transport);
+
+ return $copy;
+ }
+
+ /** @internal Freeze local diagnostic values before workflow code branches on them.
+ * @param array