diff --git a/.github/workflows/published-worker-metrics.yml b/.github/workflows/published-worker-metrics.yml index ac00f952..a91b58d7 100644 --- a/.github/workflows/published-worker-metrics.yml +++ b/.github/workflows/published-worker-metrics.yml @@ -2,10 +2,10 @@ name: Published worker metrics on: pull_request: - paths: ['.github/workflows/published-worker-metrics.yml', 'tests/smoke/worker-metrics.php', 'tests/smoke/update-outcome.php', 'tests/smoke/workflow-completion-limits.php', 'tests/smoke/query-results.php', 'tests/ci/test_published_query_tuple.py', 'tests/ci/check_published_laravel.py', 'tests/ci/test_published_laravel.py', 'tests/smoke/activity-retry-identity.php', 'tests/smoke/activity-retry-identity-assertions.php', 'tests/PublishedActivityRetryIdentityTest.php', 'tests/ci/test_published_activity_identity_tuple.py'] + paths: ['.github/workflows/published-worker-metrics.yml', 'tests/smoke/worker-metrics.php', 'tests/smoke/update-outcome.php', 'tests/smoke/workflow-completion-limits.php', 'tests/smoke/query-results.php', 'tests/ci/test_published_query_tuple.py', 'tests/ci/check_published_laravel.py', 'tests/ci/test_published_laravel.py', 'tests/smoke/activity-retry-identity.php', 'tests/smoke/activity-retry-identity-assertions.php', 'tests/PublishedActivityRetryIdentityTest.php', 'tests/ci/test_published_activity_identity_tuple.py', 'tests/smoke/local-backoff.php', 'tests/smoke/local-backoff-assertions.php', 'tests/PublishedLocalBackoffTest.php', 'tests/ci/test_published_local_backoff_tuple.py', 'tests/fixtures/replay-regressions/local-activity-recorded-backoff-cold-replay.json'] push: branches: [main] - paths: ['.github/workflows/published-worker-metrics.yml', 'tests/smoke/worker-metrics.php', 'tests/smoke/update-outcome.php', 'tests/smoke/workflow-completion-limits.php', 'tests/smoke/query-results.php', 'tests/ci/test_published_query_tuple.py', 'tests/ci/check_published_laravel.py', 'tests/ci/test_published_laravel.py', 'tests/smoke/activity-retry-identity.php', 'tests/smoke/activity-retry-identity-assertions.php', 'tests/PublishedActivityRetryIdentityTest.php', 'tests/ci/test_published_activity_identity_tuple.py'] + paths: ['.github/workflows/published-worker-metrics.yml', 'tests/smoke/worker-metrics.php', 'tests/smoke/update-outcome.php', 'tests/smoke/workflow-completion-limits.php', 'tests/smoke/query-results.php', 'tests/ci/test_published_query_tuple.py', 'tests/ci/check_published_laravel.py', 'tests/ci/test_published_laravel.py', 'tests/smoke/activity-retry-identity.php', 'tests/smoke/activity-retry-identity-assertions.php', 'tests/PublishedActivityRetryIdentityTest.php', 'tests/ci/test_published_activity_identity_tuple.py', 'tests/smoke/local-backoff.php', 'tests/smoke/local-backoff-assertions.php', 'tests/PublishedLocalBackoffTest.php', 'tests/ci/test_published_local_backoff_tuple.py', 'tests/fixtures/replay-regressions/local-activity-recorded-backoff-cold-replay.json'] workflow_dispatch: inputs: sdk_version: @@ -24,6 +24,10 @@ on: description: Qualify ordinary retry IDs on published SDK 3.18.2/native 3.8.0 type: boolean default: false + local_backoff: + description: Qualify live local retry delays and separate recorded-history replay + type: boolean + default: false laravel_bridges: description: Qualify registry SDK installation on both patched Laravel floors type: boolean @@ -64,15 +68,31 @@ jobs: COMPLETION_LIMITS: ${{ inputs.completion_limits }} QUERY_RESULTS: ${{ inputs.query_results }} LARAVEL_BRIDGES: ${{ inputs.laravel_bridges }} + LOCAL_BACKOFF: ${{ inputs.local_backoff }} + run: | + [[ "$GITHUB_EVENT_NAME" = workflow_dispatch && "$GITHUB_REF" = refs/heads/main ]] + [[ -z "$CANDIDATE_SOURCE" && "$COMPLETION_LIMITS" != true && "$QUERY_RESULTS" != true && "$LARAVEL_BRIDGES" != true && "$LOCAL_BACKOFF" != true ]] + [[ "$SDK_VERSION" = 3.18.2 && "$SERVER_VERSION" = 3.8.0 ]] + [[ "$SERVER_IMAGE" = durableworkflow/server@sha256:7988a23525d9e50e997a9f6f02e801300ed4f626de3cf6977033657ae7e9f58e ]] + - name: Require the published local backoff tuple + if: inputs.local_backoff + env: + CANDIDATE_SOURCE: ${{ inputs.candidate_source }} + COMPLETION_LIMITS: ${{ inputs.completion_limits }} + QUERY_RESULTS: ${{ inputs.query_results }} + LARAVEL_BRIDGES: ${{ inputs.laravel_bridges }} + ACTIVITY_RETRY_IDENTITY: ${{ inputs.activity_retry_identity }} + PAGINATION: ${{ inputs.pagination }} run: | [[ "$GITHUB_EVENT_NAME" = workflow_dispatch && "$GITHUB_REF" = refs/heads/main ]] - [[ -z "$CANDIDATE_SOURCE" && "$COMPLETION_LIMITS" != true && "$QUERY_RESULTS" != true && "$LARAVEL_BRIDGES" != true ]] + [[ -z "$CANDIDATE_SOURCE" && "$COMPLETION_LIMITS" != true && "$QUERY_RESULTS" != true && "$LARAVEL_BRIDGES" != true && "$ACTIVITY_RETRY_IDENTITY" != true && "$PAGINATION" != true ]] [[ "$SDK_VERSION" = 3.18.2 && "$SERVER_VERSION" = 3.8.0 ]] [[ "$SERVER_IMAGE" = durableworkflow/server@sha256:7988a23525d9e50e997a9f6f02e801300ed4f626de3cf6977033657ae7e9f58e ]] - name: Require the exact published query tuple selectors run: | python3 -I -S -B tests/ci/test_published_query_tuple.py python3 -I -S -B tests/ci/test_published_activity_identity_tuple.py + python3 -I -S -B tests/ci/test_published_local_backoff_tuple.py python3 -S -B -m unittest discover -s tests/ci -p test_published_laravel.py - name: Create isolated Composer state run: | @@ -192,6 +212,22 @@ jobs: run: | php -l tests/smoke/activity-retry-identity.php timeout 60s php tests/smoke/activity-retry-identity.php "$CONSUMER/vendor/autoload.php" | tee activity-retry-identity.json + - name: Assert installed local retry delays and recorded-history compatibility + if: inputs.local_backoff + run: | + php -l tests/smoke/local-backoff.php + timeout 90s php tests/smoke/local-backoff.php "$CONSUMER/vendor/autoload.php" | tee local-backoff.json + php -r '$r=json_decode(file_get_contents("local-backoff.json"),true,flags:JSON_THROW_ON_ERROR); + if (($r["sdk_version"]??null)!==getenv("SDK_VERSION") || ($r["server_version"]??null)!==getenv("SERVER_VERSION") + || ($r["source_fallback"]??null)!==false || ($r["legacy_replay_passed"]??null)!==true + || ($r["legacy_replay_callback_count"]??null)!==0 || array_keys($r["cases"]??[])!==["short","empty"]) { + throw new RuntimeException("Missing exact installed local-backoff evidence."); + } + foreach ($r["cases"] as $case) { + if (($case["workflow_completed"]??null)!==true || ($case["worker_deregistered"]??null)!==true || count($case["callbacks"]??[])!==5) { + throw new RuntimeException("Missing completed local retry case."); + } + }' - name: Retain the tuple, observations and installed dependency graph if: always() uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 @@ -203,6 +239,7 @@ jobs: workflow-completion-limits.json query-results.json activity-retry-identity.json + local-backoff.json candidate-package.json ${{ env.CONSUMER }}/candidate.zip ${{ env.CONSUMER }}/composer.lock @@ -215,7 +252,7 @@ jobs: run: docker rm --force --volumes sdk-php-metrics-runtime installed-laravel-bridges: - if: github.event_name == 'workflow_dispatch' && inputs.laravel_bridges && !inputs.activity_retry_identity + if: github.event_name == 'workflow_dispatch' && inputs.laravel_bridges && !inputs.activity_retry_identity && !inputs.local_backoff runs-on: ubuntu-24.04 timeout-minutes: 15 strategy: diff --git a/tests/PublishedLocalBackoffTest.php b/tests/PublishedLocalBackoffTest.php new file mode 100644 index 00000000..0c521025 --- /dev/null +++ b/tests/PublishedLocalBackoffTest.php @@ -0,0 +1,90 @@ + $delays + * @return array{list>, list>} + */ + private static function evidence(array $delays): array + { + $callbacks = []; + $history = [['event_type' => 'ActivityScheduled']]; + $time = 1_000_000_000; + foreach (range(1, count($delays) + 1) as $number) { + $callbacks[] = ['workflow_id' => 'workflow', 'run_id' => 'run', 'task_id' => 'workflow-task', + 'activity_attempt_id' => 'worker-'.$number, 'attempt_number' => $number, 'monotonic_ns' => $time]; + $payload = ['execution_mode' => 'local', 'activity_execution_id' => 'execution', + 'workflow_task_id' => 'workflow-task', 'worker_attempt_id' => 'worker-'.$number, + 'activity_attempt_id' => 'native-'.$number, 'attempt_number' => $number]; + $history[] = ['event_type' => 'ActivityStarted', 'payload' => $payload]; + if (array_key_exists($number - 1, $delays)) { + $history[] = ['event_type' => 'ActivityRetryScheduled', 'payload' => $payload + [ + 'retry_after_attempt_id' => 'native-'.$number, 'retry_after_attempt' => $number, + 'retry_backoff_seconds' => $delays[$number - 1]]]; + $time += $delays[$number - 1] * 1_000_000_000 + 1_000_000; + } else { + $history[] = ['event_type' => 'ActivityCompleted', 'payload' => $payload]; + } + } + $history[] = ['event_type' => 'WorkflowCompleted']; + + return [$callbacks, $history]; + } + + public function testRealDelayAndDurableReceiptMustAgree(): void + { + foreach ([[1, 2, 2, 2], [0, 0, 0, 0]] as $delays) { + [$callbacks, $history] = self::evidence($delays); + self::assertCount(4, verifyLocalBackoff($callbacks, 'workflow', 'run', $history, $delays)); + } + } + + public function testZeroAfterTheListEndsRefuses(): void + { + [$callbacks, $history] = self::evidence([1, 2, 0, 0]); + $this->expectException(RuntimeException::class); + verifyLocalBackoff($callbacks, 'workflow', 'run', $history, [1, 2, 2, 2]); + } + + public function testRecordedDelayCannotSubstituteForElapsedTime(): void + { + [$callbacks, $history] = self::evidence([1, 2, 2, 2]); + $callbacks[3]['monotonic_ns'] = $callbacks[2]['monotonic_ns'] + 1_000_000; + $this->expectException(RuntimeException::class); + verifyLocalBackoff($callbacks, 'workflow', 'run', $history, [1, 2, 2, 2]); + } + + public function testForeignAttemptsAndRunCannotQualifyTiming(): void + { + foreach (['activity_attempt_id', 'workflow_id', 'run_id'] as $field) { + [$callbacks, $history] = self::evidence([1, 2, 2, 2]); + $callbacks[4][$field] = 'foreign'; + try { + verifyLocalBackoff($callbacks, 'workflow', 'run', $history, [1, 2, 2, 2]); + self::fail('Accepted unrelated '.$field); + } catch (RuntimeException $error) { + self::assertNotSame('', $error->getMessage()); + } + } + } + + public function testEmptyPolicyCannotAddAnUndeclaredOneSecondDelay(): void + { + [$callbacks, $history] = self::evidence([0, 0, 0, 0]); + $callbacks[1]['monotonic_ns'] += 1_000_000_000; + $this->expectException(RuntimeException::class); + verifyLocalBackoff($callbacks, 'workflow', 'run', $history, [0, 0, 0, 0]); + } +} diff --git a/tests/Support/ReplayRegressionFixture.php b/tests/Support/ReplayRegressionFixture.php index ccbf5bee..f370645a 100644 --- a/tests/Support/ReplayRegressionFixture.php +++ b/tests/Support/ReplayRegressionFixture.php @@ -211,6 +211,7 @@ public static function execute(array $fixture): array $localActivityInvocations = 0; $localActivityExecutor = in_array($workflowType, [ self::BOUNDED_LOCAL_FAILURE, 'golden.local-activity-terminal-failure', 'golden.local-activity-recovered', 'golden.prepared-local-cleanup', + 'golden.local-activity-backoff-legacy', 'golden.prepared-local-group', 'golden.prepared-local-scoped-cleanup', ], true) ? static function () use (&$localActivityInvocations, $workflowType): array { @@ -777,6 +778,16 @@ private static function workflow(string $workflowType): callable return $origin === null ? 'unscheduled' : $origin->scheduleId.'#'.$origin->triggerNumber.'@'.$origin->occurrenceTime?->format(DATE_ATOM); }, + 'golden.local-activity-backoff-legacy' => static function (WorkflowContext $context): string { + try { + $context->localActivity('golden.local-unavailable', [], + ['retry_policy' => ['max_attempts' => 4, 'backoff_seconds' => [1]]]); + } catch (ActivityFailed) { + return 'retries-exhausted'; + } + + return 'unexpected-success'; + }, 'golden.local-activity-backoff-repeat' => static function (WorkflowContext $context): string { // .github#219: five attempts with backoff [1, 2] wait 1, 2, 2, 2 between them. try { diff --git a/tests/ci/test_published_activity_identity_tuple.py b/tests/ci/test_published_activity_identity_tuple.py index abdffd4c..24d1b612 100644 --- a/tests/ci/test_published_activity_identity_tuple.py +++ b/tests/ci/test_published_activity_identity_tuple.py @@ -13,7 +13,7 @@ def guard_source(): def require_disjoint_jobs(source): - expected = " if: github.event_name == 'workflow_dispatch' && inputs.laravel_bridges && !inputs.activity_retry_identity" + expected = " if: github.event_name == 'workflow_dispatch' && inputs.laravel_bridges && !inputs.activity_retry_identity && !inputs.local_backoff" header = source.split(" installed-laravel-bridges:\n", 1)[1].split(" runs-on:", 1)[0] if header.strip() != expected.strip(): raise ValueError("The sibling Laravel job must not start for activity identity qualification") @@ -25,7 +25,7 @@ def setUp(self): "PATH": os.defpath, "GITHUB_EVENT_NAME": "workflow_dispatch", "GITHUB_REF": "refs/heads/main", "SDK_VERSION": "3.18.2", "SERVER_VERSION": "3.8.0", "SERVER_IMAGE": "durableworkflow/server@sha256:7988a23525d9e50e997a9f6f02e801300ed4f626de3cf6977033657ae7e9f58e", - "CANDIDATE_SOURCE": "", "COMPLETION_LIMITS": "false", "QUERY_RESULTS": "false", "LARAVEL_BRIDGES": "false", + "CANDIDATE_SOURCE": "", "COMPLETION_LIMITS": "false", "QUERY_RESULTS": "false", "LARAVEL_BRIDGES": "false", "LOCAL_BACKOFF": "false", } def accepts(self, changes=None): @@ -38,7 +38,7 @@ def test_exact_published_tuple(self): def test_source_selector_and_unrelated_qualification_flags_refuse(self): for variable, value in [("CANDIDATE_SOURCE", "a" * 40), ("COMPLETION_LIMITS", "true"), - ("QUERY_RESULTS", "true"), ("LARAVEL_BRIDGES", "true")]: + ("QUERY_RESULTS", "true"), ("LARAVEL_BRIDGES", "true"), ("LOCAL_BACKOFF", "true")]: with self.subTest(variable=variable): self.assertFalse(self.accepts({variable: value})) diff --git a/tests/ci/test_published_local_backoff_tuple.py b/tests/ci/test_published_local_backoff_tuple.py new file mode 100644 index 00000000..b6a56cd5 --- /dev/null +++ b/tests/ci/test_published_local_backoff_tuple.py @@ -0,0 +1,64 @@ +"""Run the local-backoff selectors without installing or executing an SDK.""" + +import os +from pathlib import Path +import subprocess +import unittest + + +def source(): + return (Path(__file__).resolve().parents[2] / ".github/workflows/published-worker-metrics.yml").read_text() + + +def guard(): + step = source().split(" - name: Require the published local backoff tuple\n", 1)[1].split(" - ", 1)[0] + return "\n".join(line[10:] for line in step.split(" run: |\n", 1)[1].splitlines()) + + +def disjoint_jobs(workflow): + header = workflow.split(" installed-laravel-bridges:\n", 1)[1].split(" runs-on:", 1)[0] + if " && !inputs.local_backoff" not in header: + raise ValueError("Local-backoff qualification must exclude the sibling Laravel install job") + + +class PublishedLocalBackoffTupleTest(unittest.TestCase): + def setUp(self): + self.env = { + "PATH": os.defpath, "GITHUB_EVENT_NAME": "workflow_dispatch", "GITHUB_REF": "refs/heads/main", + "SDK_VERSION": "3.18.2", "SERVER_VERSION": "3.8.0", + "SERVER_IMAGE": "durableworkflow/server@sha256:7988a23525d9e50e997a9f6f02e801300ed4f626de3cf6977033657ae7e9f58e", + "CANDIDATE_SOURCE": "", "COMPLETION_LIMITS": "false", "QUERY_RESULTS": "false", + "LARAVEL_BRIDGES": "false", "ACTIVITY_RETRY_IDENTITY": "false", "PAGINATION": "false", + } + + def accepts(self, changes=None): + result = subprocess.run(["bash", "--noprofile", "--norc", "-e", "-o", "pipefail", "-c", guard()], + env=self.env | (changes or {}), capture_output=True, check=False) + return result.returncode == 0 + + def test_exact_published_tuple(self): + self.assertTrue(self.accepts()) + + def test_other_flags_candidates_and_foreign_tuples_refuse(self): + for key, value in [("CANDIDATE_SOURCE", "a" * 40), ("COMPLETION_LIMITS", "true"), + ("QUERY_RESULTS", "true"), ("LARAVEL_BRIDGES", "true"), + ("ACTIVITY_RETRY_IDENTITY", "true"), ("PAGINATION", "true"), + ("SDK_VERSION", "3.18.3"), ("SERVER_VERSION", "3.9.1"), + ("SERVER_IMAGE", "durableworkflow/server:3.8.0"), + ("GITHUB_EVENT_NAME", "pull_request"), ("GITHUB_REF", "refs/heads/topic")]: + with self.subTest(key=key): + self.assertFalse(self.accepts({key: value})) + + def test_sibling_laravel_job_cannot_start_for_local_backoff(self): + disjoint_jobs(source()) + with self.assertRaises(ValueError): + disjoint_jobs(source().replace(" && !inputs.local_backoff", "")) + + def test_typed_nonempty_evidence_required(self): + self.assertIn('json_decode(file_get_contents("local-backoff.json"),true,flags:JSON_THROW_ON_ERROR)', source()) + self.assertIn('"legacy_replay_callback_count"', source()) + self.assertIn('array_keys($r["cases"]??[])!==["short","empty"]', source()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/fixtures/replay-regressions/local-activity-recorded-backoff-cold-replay.json b/tests/fixtures/replay-regressions/local-activity-recorded-backoff-cold-replay.json new file mode 100644 index 00000000..843bf597 --- /dev/null +++ b/tests/fixtures/replay-regressions/local-activity-recorded-backoff-cold-replay.json @@ -0,0 +1,32 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "php-local-activity-recorded-backoff-cold-replay", + "protocol_version": "1.19", + "bindings": ["php"], + "workflow": {"type": "golden.local-activity-backoff-legacy", "input": []}, + "history": [ + { + "event_type": "ActivityScheduled", + "payload": {"sequence": 1, "activity_type": "golden.local-unavailable", "execution_mode": "local", "retry_policy": {"max_attempts": 4, "backoff_seconds": [1]}} + }, + { + "event_type": "ActivityFailed", + "payload": { + "sequence": 1, + "activity_type": "golden.local-unavailable", + "execution_mode": "local", + "outcome": "failed", + "message": "recorded legacy failure", + "exception_type": "RuntimeException", + "attempts": [ + {"attempt_id": "legacy-1", "attempt_number": 1, "outcome": "failed", "retry_reason": "failure", "backoff_seconds": 1}, + {"attempt_id": "legacy-2", "attempt_number": 2, "outcome": "failed", "retry_reason": "failure", "backoff_seconds": 0}, + {"attempt_id": "legacy-3", "attempt_number": 3, "outcome": "failed", "retry_reason": "failure", "backoff_seconds": 0}, + {"attempt_id": "legacy-4", "attempt_number": 4, "outcome": "failed"} + ] + } + } + ], + "expected": {"command_sequence": [{"type": "complete_workflow", "result": "retries-exhausted"}]} +} diff --git a/tests/smoke/local-backoff-assertions.php b/tests/smoke/local-backoff-assertions.php new file mode 100644 index 00000000..5fb1d002 --- /dev/null +++ b/tests/smoke/local-backoff-assertions.php @@ -0,0 +1,78 @@ +> $callbacks + * @param list> $history + * @param list $expectedDelays + * @return list + */ +function verifyLocalBackoff(array $callbacks, string $workflowId, string $runId, array $history, array $expectedDelays): array +{ + $count = count($expectedDelays) + 1; + if (count($callbacks) !== $count || array_column($callbacks, 'attempt_number') !== range(1, $count)) { + throw new RuntimeException('Require exactly the declared local callback attempts.'); + } + $events = static fn (string $type): array => array_values(array_filter($history, + static fn (array $event): bool => ($event['event_type'] ?? null) === $type)); + foreach (['ActivityScheduled' => 1, 'ActivityStarted' => $count, 'ActivityRetryScheduled' => $count - 1, + 'ActivityCompleted' => 1, 'WorkflowCompleted' => 1, 'ActivityFailed' => 0, 'WorkflowFailed' => 0] as $type => $expected) { + if (count($events($type)) !== $expected) { + throw new RuntimeException('Unexpected local history count for '.$type); + } + } + $started = $events('ActivityStarted'); + $retries = $events('ActivityRetryScheduled'); + $executionId = $started[0]['payload']['activity_execution_id'] ?? null; + if (!is_string($executionId) || $executionId === '' + || count(array_unique(array_column($callbacks, 'activity_attempt_id'))) !== $count + || count(array_unique(array_map(static fn (array $event): mixed => $event['payload']['activity_attempt_id'] ?? null, $started))) !== $count) { + throw new RuntimeException('Require one durable local execution and distinct worker attempt identities.'); + } + $gaps = []; + foreach ($callbacks as $index => $callback) { + $payload = $started[$index]['payload'] ?? []; + if (($callback['workflow_id'] ?? null) !== $workflowId || ($callback['run_id'] ?? null) !== $runId + || !is_string($callback['task_id'] ?? null) || $callback['task_id'] === '' + || !is_string($callback['activity_attempt_id'] ?? null) || $callback['activity_attempt_id'] === '' + || !is_int($callback['monotonic_ns'] ?? null) + || ($payload['execution_mode'] ?? null) !== 'local' + || ($payload['activity_execution_id'] ?? null) !== $executionId + || ($payload['workflow_task_id'] ?? null) !== $callback['task_id'] + || ($payload['worker_attempt_id'] ?? null) !== $callback['activity_attempt_id'] + || !is_string($payload['activity_attempt_id'] ?? null) || $payload['activity_attempt_id'] === '' + || ($payload['attempt_number'] ?? null) !== $index + 1) { + throw new RuntimeException('Each local callback must bind its exact durable started identity.'); + } + if ($index === $count - 1) { + continue; + } + $retry = $retries[$index]['payload'] ?? []; + if (($retry['execution_mode'] ?? null) !== 'local' + || ($retry['activity_execution_id'] ?? null) !== $executionId + || ($retry['worker_attempt_id'] ?? null) !== $callback['activity_attempt_id'] + || ($retry['retry_after_attempt_id'] ?? null) !== ($payload['activity_attempt_id'] ?? null) + || ($retry['retry_after_attempt'] ?? null) !== $index + 1 + || ($retry['retry_backoff_seconds'] ?? null) !== $expectedDelays[$index]) { + throw new RuntimeException('The durable local retry must preserve the actual attempt and exact delay.'); + } + $gap = ($callbacks[$index + 1]['monotonic_ns'] - $callback['monotonic_ns']) / 1_000_000_000; + if ($gap < $expectedDelays[$index] || $gap > ($expectedDelays[$index] === 0 ? 0.75 : 10)) { + throw new RuntimeException('The real callback gap must include its declared delay within the finite budget.'); + } + $gaps[] = $gap; + } + $terminal = $events('ActivityCompleted')[0]['payload'] ?? []; + if (($terminal['execution_mode'] ?? null) !== 'local' || ($terminal['activity_execution_id'] ?? null) !== $executionId + || ($terminal['worker_attempt_id'] ?? null) !== $callbacks[$count - 1]['activity_attempt_id'] + || ($terminal['attempt_number'] ?? null) !== $count) { + throw new RuntimeException('The terminal local result must belong to the final observed attempt.'); + } + + return $gaps; +} diff --git a/tests/smoke/local-backoff.php b/tests/smoke/local-backoff.php new file mode 100644 index 00000000..e7d54485 --- /dev/null +++ b/tests/smoke/local-backoff.php @@ -0,0 +1,133 @@ +getFileName()); +if ($version !== '3.18.2' || getenv('SDK_VERSION') !== $version + || $reference !== 'f9483bb2b5b45f591a82d9e5419c5b60a6e967c6' + || $root === false || $implementation === false || !str_starts_with($implementation, $root.'/') + || getenv('SERVER_VERSION') !== '3.8.0' + || getenv('SERVER_IMAGE') !== 'durableworkflow/server@sha256:7988a23525d9e50e997a9f6f02e801300ed4f626de3cf6977033657ae7e9f58e') { + throw new RuntimeException('Require the exact published SDK 3.18.2/native 3.8.0 tuple.'); +} +$installed = json_decode(file_get_contents(dirname($argv[1]).'/composer/installed.json'), true, flags: JSON_THROW_ON_ERROR); +$packages = array_values(array_filter($installed['packages'] ?? [], static fn (array $row): bool => $row['name'] === 'durable-workflow/sdk')); +if (count($packages) !== 1 || ($packages[0]['installation-source'] ?? null) !== 'dist' + || ($packages[0]['source']['reference'] ?? null) !== $reference + || ($packages[0]['dist']['reference'] ?? null) !== $reference || ($packages[0]['dist']['type'] ?? null) !== 'zip') { + throw new RuntimeException('Require the exact registry ZIP installation without source fallback.'); +} +$client = new Client(getenv('METRICS_RUNTIME_URL'), token: getenv('METRICS_TOKEN')); +$info = $client->clusterInfo(); +if ($info->version !== '3.8.0' || ($info->raw['implementation'] ?? null) !== 'rust') { + throw new RuntimeException('Require the selected published native producer.'); +} +$cases = []; +foreach (['short' => [1, 2], 'empty' => []] as $name => $backoff) { + $queue = 'published-local-backoff-'.bin2hex(random_bytes(8)); + $callbacks = []; + $worker = new Worker($client, $queue); + $worker->registerActivity('backoff.local', static function (ActivityContext $context) use (&$callbacks): string { + $callbacks[] = [ + 'workflow_id' => $context->workflowId, 'run_id' => $context->runId, + 'task_id' => $context->taskId, 'activity_attempt_id' => $context->activityAttemptId, + 'attempt_number' => $context->attemptNumber, 'monotonic_ns' => hrtime(true), + ]; + if ($context->attemptNumber < 5) { + throw new RuntimeException('Expected local retry failure.'); + } + if ($context->attemptNumber !== 5) { + throw new RuntimeException('Unexpected extra local attempt.'); + } + + return 'local-retry-complete'; + }); + $policy = ['max_attempts' => 5]; + if ($backoff !== []) { + $policy['backoff_seconds'] = $backoff; + } + $worker->registerWorkflow('backoff.workflow', static function (WorkflowContext $context) use ($worker, $policy): string { + $result = $context->localActivity('backoff.local', [], ['retry_policy' => $policy]); + $worker->requestShutdown(); + + return $result; + }); + $handle = $client->startWorkflow('backoff.workflow', 'local-backoff-'.bin2hex(random_bytes(8)), $queue, + executionTimeoutSeconds: 45, runTimeoutSeconds: 45); + if ($handle->selectedRunId === null) { + throw new RuntimeException('Require the exact started run.'); + } + $worker->run(0); + $history = []; + foreach ($client->iterateWorkflowHistory($handle->workflowId, $handle->selectedRunId, pageSize: 100) as $event) { + $history[] = $event; + if (count($history) > 64) { + throw new RuntimeException('The single local activity exceeded the bounded history.'); + } + } + $expected = $name === 'short' ? [1, 2, 2, 2] : [0, 0, 0, 0]; + $gaps = verifyLocalBackoff($callbacks, $handle->workflowId, $handle->selectedRunId, $history, $expected); + $scheduled = array_values(array_filter($history, static fn (array $event): bool => $event['event_type'] === 'ActivityScheduled')); + if (($scheduled[0]['payload']['activity']['retry_policy'] ?? null) != $policy) { + throw new RuntimeException('The producer must retain the original explicit-list policy without a generated maximum.'); + } + if ($handle->resultOfSelectedRun(timeoutSeconds: 5) !== 'local-retry-complete' + || $handle->describeSelectedRun()->status !== 'completed' + || ($client->listWorkers($queue)['workers'] ?? null) !== []) { + throw new RuntimeException('The exact run must finish and its managed Worker must deregister.'); + } + $cases[$name] = ['workflow_id' => $handle->workflowId, 'run_id' => $handle->selectedRunId, + 'policy' => $policy, 'expected_delays_seconds' => $expected, 'callback_gaps_seconds' => $gaps, + 'callbacks' => $callbacks, 'history' => $history, 'workflow_completed' => true, 'worker_deregistered' => true]; +} + +// Synthetic recorded legacy history is a separate installed-Replayer check, +// not a database seeded by a previous SDK or an actual older-producer run. +$fixturePath = __DIR__.'/../fixtures/replay-regressions/local-activity-recorded-backoff-cold-replay.json'; +$fixtureBytes = file_get_contents($fixturePath); +$fixture = json_decode($fixtureBytes, true, flags: JSON_THROW_ON_ERROR); +$codec = new AvroPayloadCodec(); +$calls = 0; +$legacyWorkflow = static function (WorkflowContext $context): string { + try { + $context->localActivity('golden.local-unavailable', [], + ['retry_policy' => ['max_attempts' => 4, 'backoff_seconds' => [1]]]); + } catch (ActivityFailed) { + return 'retries-exhausted'; + } + + throw new RuntimeException('The recorded legacy failure must remain authoritative.'); +}; +$commands = (new Replayer($codec))->replay($legacyWorkflow, $fixture['history'], [], 'legacy-replay', + localActivityExecutor: static function () use (&$calls): never { ++$calls; throw new RuntimeException('Recorded local activity must not execute again.'); })->commands; +if ($calls !== 0 || count($commands) !== 1 || $commands[0]['type'] !== 'complete_workflow' + || $codec->decodeEnvelope($commands[0]['result']) !== 'retries-exhausted' + || array_column($fixture['history'][1]['payload']['attempts'], 'backoff_seconds') !== [1, 0, 0] + || file_get_contents($fixturePath) !== $fixtureBytes) { + throw new RuntimeException('Legacy recorded delays must replay without local calls or rewritten history.'); +} +echo json_encode([ + 'sdk_version' => $version, 'sdk_reference' => $reference, + 'server_version' => $info->version, 'server_image' => getenv('SERVER_IMAGE'), + 'runner_commit' => getenv('GITHUB_SHA') ?: null, 'cases' => $cases, + 'legacy_scope' => 'synthetic_recorded_history_installed_replayer', + 'legacy_fixture_sha256' => hash('sha256', $fixtureBytes), 'legacy_recorded_delays' => [1, 0, 0], + 'legacy_replay_callback_count' => $calls, 'legacy_replay_passed' => true, 'source_fallback' => false, +], JSON_THROW_ON_ERROR | JSON_PRETTY_PRINT | JSON_UNESCAPED_SLASHES).PHP_EOL;