diff --git a/.github/workflows/published-worker-metrics.yml b/.github/workflows/published-worker-metrics.yml index 26195967..ac00f952 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'] + 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'] 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'] + 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'] workflow_dispatch: inputs: sdk_version: @@ -20,6 +20,10 @@ on: description: Qualify binary and JSON query results with the installed Worker type: boolean default: false + activity_retry_identity: + description: Qualify ordinary retry IDs on published SDK 3.18.2/native 3.8.0 + type: boolean + default: false laravel_bridges: description: Qualify registry SDK installation on both patched Laravel floors type: boolean @@ -53,9 +57,22 @@ jobs: - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6 with: persist-credentials: false + - name: Require the published activity identity tuple + if: inputs.activity_retry_identity + env: + CANDIDATE_SOURCE: ${{ inputs.candidate_source }} + COMPLETION_LIMITS: ${{ inputs.completion_limits }} + QUERY_RESULTS: ${{ inputs.query_results }} + LARAVEL_BRIDGES: ${{ inputs.laravel_bridges }} + run: | + [[ "$GITHUB_EVENT_NAME" = workflow_dispatch && "$GITHUB_REF" = refs/heads/main ]] + [[ -z "$CANDIDATE_SOURCE" && "$COMPLETION_LIMITS" != true && "$QUERY_RESULTS" != true && "$LARAVEL_BRIDGES" != 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 -S -B -m unittest discover -s tests/ci -p test_published_laravel.py - name: Create isolated Composer state run: | @@ -170,6 +187,11 @@ jobs: if (($r["sdk_version"]??null)!==getenv("SDK_VERSION") || ($r["server_version"]??null)!==getenv("SERVER_VERSION")) { throw new RuntimeException("Query evidence tuple mismatch."); }' + - name: Assert ordinary activity retry identity over real HTTP + if: inputs.activity_retry_identity + 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: Retain the tuple, observations and installed dependency graph if: always() uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 @@ -180,6 +202,7 @@ jobs: update-outcome.json workflow-completion-limits.json query-results.json + activity-retry-identity.json candidate-package.json ${{ env.CONSUMER }}/candidate.zip ${{ env.CONSUMER }}/composer.lock @@ -192,7 +215,7 @@ jobs: run: docker rm --force --volumes sdk-php-metrics-runtime installed-laravel-bridges: - if: github.event_name == 'workflow_dispatch' && inputs.laravel_bridges + if: github.event_name == 'workflow_dispatch' && inputs.laravel_bridges && !inputs.activity_retry_identity runs-on: ubuntu-24.04 timeout-minutes: 15 strategy: diff --git a/tests/PublishedActivityRetryIdentityTest.php b/tests/PublishedActivityRetryIdentityTest.php new file mode 100644 index 00000000..1706fe7b --- /dev/null +++ b/tests/PublishedActivityRetryIdentityTest.php @@ -0,0 +1,85 @@ +>, list>} */ + private static function evidence(): array + { + $callbacks = []; + foreach ([1, 2] as $attempt) { + $callbacks[] = ['activity_execution_id' => 'execution', 'workflow_id' => 'workflow', 'run_id' => 'run', + 'task_id' => 'task-'.$attempt, 'activity_attempt_id' => 'attempt-'.$attempt, 'attempt_number' => $attempt]; + } + $started = static fn (array $callback): array => ['event_type' => 'ActivityStarted', 'payload' => [ + 'activity_execution_id' => $callback['activity_execution_id'], 'task' => ['id' => $callback['task_id']], + 'activity_attempt_id' => $callback['activity_attempt_id'], 'attempt_number' => $callback['attempt_number'], + ]]; + $history = [ + ['event_type' => 'ActivityScheduled'], $started($callbacks[0]), + ['event_type' => 'ActivityRetryScheduled', 'payload' => [ + 'activity_execution_id' => 'execution', 'retry_of_task_id' => 'task-1', 'retry_task_id' => 'task-2', + 'retry_after_attempt_id' => 'attempt-1', 'retry_after_attempt' => 1, 'max_attempts' => 2, 'retry_backoff_seconds' => 0, + ]], + $started($callbacks[1]), ['event_type' => 'ActivityCompleted'], ['event_type' => 'WorkflowCompleted'], + ]; + + return [$callbacks, $history]; + } + + public function testBoundCallbackPairAndDurableRetryReceipt(): void + { + [$callbacks, $history] = self::evidence(); + verifyActivityRetryIdentity($callbacks, 'workflow', 'run', $history); + self::assertCount(2, $callbacks); + } + + public function testMissingLogicalIdentityCannotBorrowTaskIdentity(): void + { + [$callbacks, $history] = self::evidence(); + $callbacks[0]['activity_execution_id'] = null; + $this->expectException(RuntimeException::class); + verifyActivityRetryIdentity($callbacks, 'workflow', 'run', $history); + } + + public function testChangedLogicalIdentityAndReusedAttemptIdentitiesRefuse(): void + { + foreach (['activity_execution_id', 'workflow_id', 'run_id', 'task_id', 'activity_attempt_id', 'attempt_number'] as $field) { + [$callbacks, $history] = self::evidence(); + $callbacks[1][$field] = in_array($field, ['task_id', 'activity_attempt_id', 'attempt_number'], true) + ? $callbacks[0][$field] : 'foreign'; + try { + verifyActivityRetryIdentity($callbacks, 'workflow', 'run', $history); + self::fail('Accepted changed or reused '.$field); + } catch (RuntimeException $error) { + self::assertNotSame('', $error->getMessage()); + } + } + } + + public function testHistoryFromAnotherAttemptCannotQualifyTheCallbackPair(): void + { + [$callbacks, $history] = self::evidence(); + $history[2]['payload']['retry_task_id'] = 'foreign-task'; + $this->expectException(RuntimeException::class); + verifyActivityRetryIdentity($callbacks, 'workflow', 'run', $history); + } + + public function testInventedSecondAttemptIdentityRefuses(): void + { + [$callbacks, $history] = self::evidence(); + $callbacks[1]['activity_attempt_id'] = 'unrelated-nonempty-attempt'; + $this->expectException(RuntimeException::class); + verifyActivityRetryIdentity($callbacks, 'workflow', 'run', $history); + } +} diff --git a/tests/ci/test_published_activity_identity_tuple.py b/tests/ci/test_published_activity_identity_tuple.py new file mode 100644 index 00000000..abdffd4c --- /dev/null +++ b/tests/ci/test_published_activity_identity_tuple.py @@ -0,0 +1,60 @@ +"""Exercise the real pre-install selector guard without SDK or Docker execution.""" + +import os +from pathlib import Path +import subprocess +import unittest + + +def guard_source(): + workflow = (Path(__file__).resolve().parents[2] / ".github/workflows/published-worker-metrics.yml").read_text() + step = workflow.split(" - name: Require the published activity identity tuple\n", 1)[1].split(" - ", 1)[0] + return "\n".join(line[10:] for line in step.split(" run: |\n", 1)[1].splitlines()) + + +def require_disjoint_jobs(source): + expected = " if: github.event_name == 'workflow_dispatch' && inputs.laravel_bridges && !inputs.activity_retry_identity" + 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") + + +class PublishedActivityIdentityTupleTest(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", + } + + def accepts(self, changes=None): + result = subprocess.run(["bash", "--noprofile", "--norc", "-e", "-o", "pipefail", "-c", guard_source()], + 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_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")]: + with self.subTest(variable=variable): + self.assertFalse(self.accepts({variable: value})) + + def test_wrong_package_producer_and_dispatch_scope_refuse(self): + for variable, value in [("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(variable=variable): + self.assertFalse(self.accepts({variable: value})) + + def test_sibling_job_cannot_start_on_conflicting_flags(self): + source = (Path(__file__).resolve().parents[2] / ".github/workflows/published-worker-metrics.yml").read_text() + require_disjoint_jobs(source) + with self.assertRaises(ValueError): + require_disjoint_jobs(source.replace(" && !inputs.activity_retry_identity", "")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/smoke/activity-retry-identity-assertions.php b/tests/smoke/activity-retry-identity-assertions.php new file mode 100644 index 00000000..188ccbdd --- /dev/null +++ b/tests/smoke/activity-retry-identity-assertions.php @@ -0,0 +1,66 @@ +> $callbacks + * @param list> $history + */ +function verifyActivityRetryIdentity(array $callbacks, string $workflowId, string $runId, array $history): void +{ + if (count($callbacks) !== 2 || array_column($callbacks, 'attempt_number') !== [1, 2]) { + throw new RuntimeException('Require exactly the first and second real activity callbacks.'); + } + foreach ($callbacks as $callback) { + foreach (['activity_execution_id', 'workflow_id', 'run_id', 'task_id', 'activity_attempt_id'] as $field) { + if (!is_string($callback[$field] ?? null) || $callback[$field] === '') { + throw new RuntimeException('Missing received callback identity: '.$field); + } + } + if ($callback['workflow_id'] !== $workflowId || $callback['run_id'] !== $runId) { + throw new RuntimeException('The callback must belong to the exact started workflow and run.'); + } + } + [$first, $second] = $callbacks; + if ($first['activity_execution_id'] !== $second['activity_execution_id'] + || $first['task_id'] === $second['task_id'] + || $first['activity_attempt_id'] === $second['activity_attempt_id']) { + throw new RuntimeException('A retry must keep its logical execution and change task and attempt identity.'); + } + foreach (['ActivityScheduled' => 1, 'ActivityStarted' => 2, 'ActivityRetryScheduled' => 1, + 'ActivityCompleted' => 1, 'WorkflowCompleted' => 1, 'ActivityFailed' => 0, 'WorkflowFailed' => 0] as $type => $count) { + $events = array_values(array_filter($history, static fn (array $event): bool => ($event['event_type'] ?? null) === $type)); + if (count($events) !== $count) { + throw new RuntimeException('Unexpected real history count for '.$type); + } + if ($type !== 'ActivityRetryScheduled') { + continue; + } + $payload = $events[0]['payload'] ?? []; + foreach ([ + 'activity_execution_id' => $first['activity_execution_id'], + 'retry_of_task_id' => $first['task_id'], 'retry_task_id' => $second['task_id'], + 'retry_after_attempt_id' => $first['activity_attempt_id'], + 'retry_after_attempt' => 1, 'max_attempts' => 2, 'retry_backoff_seconds' => 0, + ] as $field => $expected) { + if (($payload[$field] ?? null) !== $expected) { + throw new RuntimeException('The durable retry receipt must bind the actual callbacks: '.$field); + } + } + } + $started = array_values(array_filter($history, + static fn (array $event): bool => ($event['event_type'] ?? null) === 'ActivityStarted')); + foreach ($callbacks as $index => $callback) { + $payload = $started[$index]['payload'] ?? []; + if (($payload['activity_execution_id'] ?? null) !== $callback['activity_execution_id'] + || ($payload['task']['id'] ?? null) !== $callback['task_id'] + || ($payload['activity_attempt_id'] ?? null) !== $callback['activity_attempt_id'] + || ($payload['attempt_number'] ?? null) !== $callback['attempt_number']) { + throw new RuntimeException('Each callback must match its durable ActivityStarted identities.'); + } + } +} diff --git a/tests/smoke/activity-retry-identity.php b/tests/smoke/activity-retry-identity.php new file mode 100644 index 00000000..b05c040d --- /dev/null +++ b/tests/smoke/activity-retry-identity.php @@ -0,0 +1,100 @@ +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 reviewed registry SDK 3.18.2/native 3.8.0 tuple without source fallback.'); +} +$lock = json_decode(file_get_contents(dirname(dirname($argv[1])).'/composer.lock'), true, flags: JSON_THROW_ON_ERROR); +$packages = array_values(array_filter($lock['packages'] ?? [], static fn (array $row): bool => $row['name'] === 'durable-workflow/sdk')); +$installed = json_decode(file_get_contents(dirname($argv[1]).'/composer/installed.json'), true, flags: JSON_THROW_ON_ERROR); +$distributions = array_values(array_filter($installed['packages'] ?? [], static fn (array $row): bool => $row['name'] === 'durable-workflow/sdk')); +if (count($packages) !== 1 || ($packages[0]['version'] ?? null) !== $version + || ($packages[0]['source']['reference'] ?? null) !== $reference + || ($packages[0]['dist']['reference'] ?? null) !== $reference + || ($packages[0]['dist']['type'] ?? null) !== 'zip' + || count($distributions) !== 1 || ($distributions[0]['installation-source'] ?? null) !== 'dist' + || ($distributions[0]['source']['reference'] ?? null) !== $reference + || ($distributions[0]['dist']['reference'] ?? null) !== $reference) { + throw new RuntimeException('The ordinary registry lock must bind the exact installed SDK distribution.'); +} + +$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.'); +} +$queue = 'published-activity-identity-'.bin2hex(random_bytes(8)); +$callbacks = []; +$worker = new Worker($client, $queue); +$worker->registerActivity('identity.retry', static function (ActivityContext $context) use (&$callbacks): string { + $callbacks[] = [ + 'activity_execution_id' => $context->activityExecutionId, + 'workflow_id' => $context->workflowId, 'run_id' => $context->runId, + 'task_id' => $context->taskId, 'activity_attempt_id' => $context->activityAttemptId, + 'attempt_number' => $context->attemptNumber, + ]; + if ($context->attemptNumber === 1) { + throw new RuntimeException('Expected first-attempt failure for the published retry identity proof.'); + } + if ($context->attemptNumber !== 2 || $context->activityExecutionId === null) { + throw new RuntimeException('Require exactly the second attempt with a received execution identity.'); + } + + return $context->activityExecutionId; +}); +$worker->registerWorkflow('identity.workflow', static function (WorkflowContext $context) use ($worker): string { + $result = $context->activity('identity.retry', [], ['retry_policy' => ['max_attempts' => 2, 'backoff_seconds' => [0]]]); + $worker->requestShutdown(); + + return $result; +}); +$handle = $client->startWorkflow('identity.workflow', 'identity-'.bin2hex(random_bytes(8)), $queue, + executionTimeoutSeconds: 45, runTimeoutSeconds: 45); +if ($handle->selectedRunId === null) { + throw new RuntimeException('The producer must return the exact started run.'); +} +$worker->run(0); +$result = $handle->resultOfSelectedRun(timeoutSeconds: 5); +$history = []; +foreach ($client->iterateWorkflowHistory($handle->workflowId, $handle->selectedRunId, pageSize: 100) as $event) { + $history[] = $event; + if (count($history) > 64) { + throw new RuntimeException('The single-activity retry history exceeded its bounded evidence size.'); + } +} +verifyActivityRetryIdentity($callbacks, $handle->workflowId, $handle->selectedRunId, $history); +if ($result !== $callbacks[1]['activity_execution_id'] + || $handle->describeSelectedRun()->status !== 'completed' + || ($client->listWorkers($queue)['workers'] ?? null) !== []) { + throw new RuntimeException('The exact run must complete with the received execution ID and deregister its Worker.'); +} +echo json_encode([ + 'sdk_version' => $version, 'sdk_reference' => $reference, + 'server_version' => $info->version, 'server_image' => getenv('SERVER_IMAGE'), + 'runner_commit' => getenv('GITHUB_SHA') ?: null, + 'started_at' => $startedAt, 'completed_at' => gmdate('c'), 'scope' => 'ordinary_remote_retry', + 'workflow_id' => $handle->workflowId, 'run_id' => $handle->selectedRunId, + 'callbacks' => $callbacks, 'history' => $history, + 'workflow_completed' => true, 'worker_deregistered' => true, 'source_fallback' => false, +], JSON_THROW_ON_ERROR | JSON_PRETTY_PRINT | JSON_UNESCAPED_SLASHES).PHP_EOL;