diff --git a/.github/workflows/phpunit-feature.yml b/.github/workflows/phpunit-feature.yml index edfd4fdd0..244ebed7b 100644 --- a/.github/workflows/phpunit-feature.yml +++ b/.github/workflows/phpunit-feature.yml @@ -190,7 +190,7 @@ jobs: run: python3 -m unittest discover -s tests/Unit/Support -p 'single_region_failover*test.py' - name: Require all PHP replay runtime cells and exact artifact identity - run: python3 -m unittest tests/Unit/Support/php_replay_published_test.py + run: python3 -m unittest tests/Unit/Support/php_replay_published_test.py tests/Unit/Support/php_condition_wait_test.py feature: name: PHPUnit feature suite diff --git a/.github/workflows/published-workflow-lifecycle.yml b/.github/workflows/published-workflow-lifecycle.yml index 457df2201..3ab8536a2 100644 --- a/.github/workflows/published-workflow-lifecycle.yml +++ b/.github/workflows/published-workflow-lifecycle.yml @@ -4,11 +4,11 @@ on: workflow_dispatch: inputs: experiment: - description: Existing lifecycle aggregate or complete PHP replay matrix + description: Lifecycle aggregate, complete PHP replay, or replay plus installed condition-wait cases required: true default: lifecycle type: choice - options: [lifecycle, php-replay] + options: [lifecycle, php-replay, php-replay-conditions] php_runner_image: description: Published tooling image digest for php-replay (no product source) required: false @@ -67,6 +67,7 @@ jobs: SERVER_PORT: '0' DW_CONFORMANCE_EXPERIMENT: ${{ inputs.experiment }} DW_REPLAY_PHP_RUNNER_IMAGE: ${{ inputs.php_runner_image }} + DW_PHP_SDK_CONFORMANCE_CONDITION_WAITS: ${{ inputs.experiment == 'php-replay-conditions' && '1' || '0' }} DW_WORKFLOW_LIFECYCLE_MYSQL_IMAGE: mysql@sha256:679e7e924f38a3cbb62a3d7df32924b83f7321a602d3f9f967c01b3df18495d6 DW_WORKFLOW_LIFECYCLE_REDIS_IMAGE: redis@sha256:0637954999d01b7c9ce9167db2da50656e2590d3b884f1c600c5f63bb6e6773c steps: @@ -84,7 +85,7 @@ jobs: done case "$DW_CONFORMANCE_EXPERIMENT" in lifecycle) ;; - php-replay) + php-replay|php-replay-conditions) [[ "$DW_REPLAY_PHP_RUNNER_IMAGE" =~ ^ghcr.io/durable-workflow/sample-app-devcontainer@sha256:[0-9a-f]{64}$ ]] ;; *) exit 2 ;; @@ -94,13 +95,17 @@ jobs: if: ${{ always() && steps.validate.outcome == 'success' }} run: | mkdir -p "$RUNNER_TEMP/published-workflow-lifecycle" - if [[ "$DW_CONFORMANCE_EXPERIMENT" == php-replay ]]; then - bash scripts/conformance/replay-published-artifacts.sh --focus php \ - --result-dir "$RUNNER_TEMP/published-workflow-lifecycle" - else - bash scripts/conformance/workflow-lifecycle-host-published-artifacts.sh \ - --result-dir "$RUNNER_TEMP/published-workflow-lifecycle" - fi + case "$DW_CONFORMANCE_EXPERIMENT" in + php-replay|php-replay-conditions) + bash scripts/conformance/replay-published-artifacts.sh --focus php \ + --result-dir "$RUNNER_TEMP/published-workflow-lifecycle" + ;; + lifecycle) + bash scripts/conformance/workflow-lifecycle-host-published-artifacts.sh \ + --result-dir "$RUNNER_TEMP/published-workflow-lifecycle" + ;; + *) exit 2 ;; + esac - name: Remove job-owned lifecycle containers and network if: always() @@ -114,7 +119,7 @@ jobs: docker compose -p "$COMPOSE_PROJECT_NAME" -f docker-compose.php.yml down --volumes --remove-orphans >/dev/null 2>&1 || true - name: Require PHP replay cleanup receipt - if: ${{ always() && steps.validate.outcome == 'success' && inputs.experiment == 'php-replay' }} + if: ${{ always() && steps.validate.outcome == 'success' && (inputs.experiment == 'php-replay' || inputs.experiment == 'php-replay-conditions') }} run: test "$(<"$RUNNER_TEMP/published-workflow-lifecycle/php-replay-cleanup.txt")" = 'cleanup=0' - name: Make lifecycle evidence readable @@ -140,6 +145,7 @@ jobs: ${{ runner.temp }}/published-workflow-lifecycle/*.txt ${{ runner.temp }}/published-workflow-lifecycle/*.csv ${{ runner.temp }}/published-workflow-lifecycle/*.tsv + ${{ runner.temp }}/published-workflow-lifecycle/*.sql ${{ runner.temp }}/published-workflow-lifecycle/*.sqlite if-no-files-found: error retention-days: 30 diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index 557a0b17f..af4f34272 100644 --- a/.github/workflows/server-parity.yml +++ b/.github/workflows/server-parity.yml @@ -14,6 +14,9 @@ on: - Cargo.lock - scripts/conformance/** - scripts/ci/** + # Publication identity is checked by Native promotion contracts. The + # fixture jobs do not read it; every other CI helper stays selected. + - '!scripts/ci/native-qualification-subject.json' - tests/Unit/ServerParity*.mjs - tests/Support/ServerParity*.mjs - scripts/conformance/server-parity* diff --git a/docker-compose.published.yml b/docker-compose.published.yml index c6eac606f..83605822c 100644 --- a/docker-compose.published.yml +++ b/docker-compose.published.yml @@ -16,7 +16,7 @@ name: durable-workflow-server # release whose default names another image). Set DW_SERVER_IMAGE to run # any other native image, by tag or digest; pin a digest for production # change control. -x-server-image: &server-image ${DW_SERVER_IMAGE:-durableworkflow/server:3.3.0} +x-server-image: &server-image ${DW_SERVER_IMAGE:-durableworkflow/server:3.3.1} services: server: diff --git a/docs/releases/3.3.1.md b/docs/releases/3.3.1.md new file mode 100644 index 000000000..bbef0ff85 --- /dev/null +++ b/docs/releases/3.3.1.md @@ -0,0 +1,51 @@ +# Durable Workflow Server 3.3.1 (native) + +Server 3.3.1 is a patch release of the native (Rust) Server, published as +`durableworkflow/server:3.3.1` and `ghcr.io/durable-workflow/server:3.3.1`. +The published Compose file defaults to it. Rolling tags move in the separate +promotion step after qualification. + +## Upgrade + +There is no schema upgrade from 3.3.0. MySQL, MariaDB and PostgreSQL remain +at native schema 11; SQLite remains at schema 10. The worker protocol stays +at `1.20`, the control plane at version `2`, and the capabilities are unchanged. +Existing clients need no SDK release for these internal query reductions. + +For upgrades from 3.2.x or earlier, follow the +[3.3.0 upgrade and rollback instructions](3.3.0.md#upgrade), including the +backup, explicit `schema-upgrade` and condition-wait rollback constraints. +This patch does not relax those constraints or repair a mixed condition-wait +chain. + +## Changes + +[PR #1099](https://github.com/durable-workflow/server/pull/1099) reduces +database statements on workflow transitions while retaining the existing +answers, transaction boundaries and lease authority: + +- Upsert a run summary instead of checking its existence before writing it. +- Read a workflow poll's cached binding and result together. +- Aggregate the projection census in one query. +- Read the authored sequence counters together, retaining integer types on + MariaDB and zero for missing rows. +- Reuse the waiting-status result already read in the transition. +- Reuse fenced worker registration within the transaction. +- Keep run tags within the existing transaction context. + +The paired poll-cache writer proposed in +[issue #1063](https://github.com/durable-workflow/server/issues/1063) is +deferred. Binding and result writes retain their separate expiry-time origins. +The other items in the performance report remain tracked individually. +No benchmark speedup or latency improvement is claimed for this release. + +The qualification tooling also verifies the published PHP replay matrix +under a read-only root, dropped capabilities and `no-new-privileges`. A private, +byte-identical PHP executable copy avoids inheriting the tooling image's file +capability; Server and SDK product bytes are not patched for the experiment. + +## Known limitations + +The [3.3.0 known limitations](3.3.0.md#known-limitations) still apply. +Source/backend query-count controls and published-artifact compatibility are +separate evidence; passing them does not reproduce the reporter's benchmark. diff --git a/scripts/conformance/condition-wait-anchored-check.sql b/scripts/conformance/condition-wait-anchored-check.sql new file mode 100644 index 000000000..b9bf6e85d --- /dev/null +++ b/scripts/conformance/condition-wait-anchored-check.sql @@ -0,0 +1,272 @@ +-- Read-only anchored check v4 for sdk-php#269 occurrence-deadline cells on +-- Native with server#1098 (condition_wait_occurrence_deadline), MySQL 8.x. +-- A distinct successor. The original condition-wait-physical-check.sql +-- (a162896d), the v1 prototype (485861da), v2 (d09159ce) and v3 (f03de3f0) +-- stay unchanged. +-- v4 changes no predicate. It makes every text value independent of the +-- client connection: each output column, each session-variable or CAST/ +-- DATE_FORMAT string, and each table-column/payload comparison is pinned to +-- utf8mb4_bin (case-sensitive, PAD SPACE). v3 aborted on a latin1 client with ERROR 1267 +-- (utf8mb3 SYSCONST @@session.time_zone UNION latin1 CAST); a utf8mb4 +-- client would instead mix utf8mb4_0900_ai_ci with the schema's +-- utf8mb4_unicode_ci. Run-ID membership uses that collation; the runner admits fixed-format ULIDs. +-- v3 checks each identity's raw JSON type is STRING before unquoting it: +-- JSON null unquotes to the text 'null' and must never pass as an identity. +-- Server-owned bytes; bind them by SHA-256. +-- +-- Native keeps the policy, original budget and anchor A in history payloads +-- (ConditionWaitOpened: deadline_policy, condition_wait_timeout_budget_seconds, +-- occurrence_deadline_at; TimerScheduled: deadline_policy, occurrence_deadline_at, +-- fire_at, timer_id) and in the first timer row's fire_at. There is no schema +-- column for them. The waits projection copies TimerScheduled.fire_at into +-- workflow_run_waits.deadline_at (wait_id = condition_wait_id). +-- +-- Every predicate refuses with "(...) IS NOT TRUE", so a missing or null +-- field is a violation, never a silent pass. +-- +-- Caller sets these, then sources this file on a read-only connection: +-- SET time_zone = '+00:00'; -- required (B-1) +-- SET @run_ids = ',...'; -- every closed run of the cell +-- SET @max_dispatch_seconds = 2; -- required, exactly the selected 2 s (B-2) +-- SET @expected_budgets = NULL; -- or a JSON object mapping +-- '|' to that chain's original T. +-- Leave NULL for a heterogeneous cell whose per-chain T is checked by +-- the SDK evidence; when set, every chain needs an equal entry and no +-- key may name a chain that is not present (B2b, B2c). +-- Output: one row per violation (assertion, run_id, detail). No rows = PASS. +-- Only SELECTs; no table or row is written. + +WITH opens AS ( + SELECT h.workflow_run_id AS run_id, h.sequence AS hseq, + JSON_UNQUOTE(JSON_EXTRACT(h.payload, '$.condition_wait_id')) AS wait_id, + JSON_UNQUOTE(JSON_EXTRACT(h.payload, '$.condition_wait_occurrence_id')) AS occ, + JSON_UNQUOTE(JSON_EXTRACT(h.payload, '$.condition_key')) AS ckey, + JSON_UNQUOTE(JSON_EXTRACT(h.payload, '$.condition_definition_fingerprint')) AS fp, + JSON_UNQUOTE(JSON_EXTRACT(h.payload, '$.deadline_policy')) AS policy, + JSON_EXTRACT(h.payload, '$.condition_wait_timeout_budget_seconds') AS budget_j, + JSON_EXTRACT(h.payload, '$.timeout_seconds') AS timeout_j, + JSON_UNQUOTE(JSON_EXTRACT(h.payload, '$.occurrence_deadline_at')) AS anchor_s, + JSON_TYPE(JSON_EXTRACT(h.payload, '$.condition_wait_id')) AS wait_t, + JSON_TYPE(JSON_EXTRACT(h.payload, '$.condition_wait_occurrence_id')) AS occ_t, + JSON_TYPE(JSON_EXTRACT(h.payload, '$.deadline_policy')) AS policy_t, + JSON_TYPE(JSON_EXTRACT(h.payload, '$.occurrence_deadline_at')) AS anchor_t + FROM workflow_history_events h + WHERE FIND_IN_SET(h.workflow_run_id, CONVERT(@run_ids USING utf8mb4) COLLATE utf8mb4_bin) AND h.event_type = 'ConditionWaitOpened' +), sched AS ( + SELECT h.workflow_run_id AS run_id, h.sequence AS hseq, + JSON_UNQUOTE(JSON_EXTRACT(h.payload, '$.condition_wait_id')) AS wait_id, + JSON_UNQUOTE(JSON_EXTRACT(h.payload, '$.timer_id')) AS timer_id, + JSON_UNQUOTE(JSON_EXTRACT(h.payload, '$.fire_at')) AS fire_s, + JSON_UNQUOTE(JSON_EXTRACT(h.payload, '$.deadline_policy')) AS policy, + JSON_UNQUOTE(JSON_EXTRACT(h.payload, '$.occurrence_deadline_at')) AS anchor_s, + JSON_TYPE(JSON_EXTRACT(h.payload, '$.condition_wait_id')) AS wait_t, + JSON_TYPE(JSON_EXTRACT(h.payload, '$.timer_id')) AS timer_t, + JSON_TYPE(JSON_EXTRACT(h.payload, '$.fire_at')) AS fire_t, + JSON_TYPE(JSON_EXTRACT(h.payload, '$.deadline_policy')) AS policy_t, + JSON_TYPE(JSON_EXTRACT(h.payload, '$.occurrence_deadline_at')) AS anchor_t + FROM workflow_history_events h + WHERE FIND_IN_SET(h.workflow_run_id, CONVERT(@run_ids USING utf8mb4) COLLATE utf8mb4_bin) AND h.event_type = 'TimerScheduled' + AND JSON_UNQUOTE(JSON_EXTRACT(h.payload, '$.timer_kind')) = 'condition_timeout' +), firsts AS ( + SELECT o.* FROM opens o + WHERE o.hseq = (SELECT MIN(f.hseq) FROM opens f WHERE f.run_id = o.run_id AND f.occ <=> o.occ) +), anchors AS ( + -- A is the first open's own timer row fire_at, read by its timer ID. The + -- bijection checks B8 make any join multiplication visible on their own. + SELECT f.run_id, f.occ, f.wait_id AS first_wait, f.budget_j, f.ckey, f.fp, f.anchor_s, + t.id AS timer_id, t.fire_at AS a_at + FROM firsts f + LEFT JOIN sched s ON s.run_id = f.run_id AND s.wait_id = f.wait_id + LEFT JOIN workflow_run_timers t ON t.id COLLATE utf8mb4_bin = s.timer_id AND t.workflow_run_id = f.run_id +) + +SELECT CONVERT('B-1 session time zone is not +00:00' USING utf8mb4) COLLATE utf8mb4_bin AS assertion, CONVERT(NULL USING utf8mb4) COLLATE utf8mb4_bin AS run_id, CONVERT(@@session.time_zone USING utf8mb4) COLLATE utf8mb4_bin AS detail +FROM DUAL WHERE (@@session.time_zone = '+00:00') IS NOT TRUE + +UNION ALL +SELECT CONVERT('B-2 dispatch bound is not the selected 2 seconds' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(NULL USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(COALESCE(CAST(@max_dispatch_seconds AS CHAR CHARACTER SET utf8mb4) COLLATE utf8mb4_bin, 'NULL') USING utf8mb4) COLLATE utf8mb4_bin +FROM DUAL WHERE (@max_dispatch_seconds <=> 2) IS NOT TRUE + +UNION ALL +SELECT CONVERT('B-3 expected budget map is not a JSON object' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(NULL USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(LEFT(CAST(@expected_budgets AS CHAR CHARACTER SET utf8mb4) COLLATE utf8mb4_bin, 200) USING utf8mb4) COLLATE utf8mb4_bin +FROM DUAL WHERE @expected_budgets IS NOT NULL + AND (JSON_VALID(@expected_budgets) AND JSON_TYPE(@expected_budgets) = 'OBJECT') IS NOT TRUE + +UNION ALL +-- Every physical open carries the complete opt-in, with integer budget and +-- remaining timeout and 0 <= timeout <= budget. +SELECT CONVERT('B1 condition wait without a complete occurrence policy' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(o.run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(COALESCE(o.wait_id, 'NULL'), ' policy=', COALESCE(o.policy, 'NULL'), ' budget=', COALESCE(CAST(o.budget_j AS CHAR CHARACTER SET utf8mb4) COLLATE utf8mb4_bin, 'NULL'), + ' timeout=', COALESCE(CAST(o.timeout_j AS CHAR CHARACTER SET utf8mb4) COLLATE utf8mb4_bin, 'NULL'), ' occ=', COALESCE(o.occ, 'NULL'), + ' anchor=', COALESCE(o.anchor_s, 'NULL')) USING utf8mb4) COLLATE utf8mb4_bin +FROM opens o +WHERE (o.policy = 'occurrence' AND JSON_TYPE(o.budget_j) = 'INTEGER' AND JSON_TYPE(o.timeout_j) = 'INTEGER' + AND CAST(o.timeout_j AS SIGNED) >= 0 AND CAST(o.timeout_j AS SIGNED) <= CAST(o.budget_j AS SIGNED) + AND o.wait_t = 'STRING' AND o.occ_t = 'STRING' AND o.policy_t = 'STRING' AND o.anchor_t = 'STRING' + AND o.wait_id <> '' AND o.occ <> '' AND o.anchor_s <> '') IS NOT TRUE + +UNION ALL +-- The chain's first open authors its whole budget. +SELECT CONVERT('B2 first open does not author its whole budget' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(f.run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(COALESCE(f.wait_id, 'NULL'), ' timeout=', COALESCE(CAST(f.timeout_j AS CHAR CHARACTER SET utf8mb4) COLLATE utf8mb4_bin, 'NULL'), + ' budget=', COALESCE(CAST(f.budget_j AS CHAR CHARACTER SET utf8mb4) COLLATE utf8mb4_bin, 'NULL')) USING utf8mb4) COLLATE utf8mb4_bin +FROM firsts f +WHERE (JSON_TYPE(f.timeout_j) = 'INTEGER' AND JSON_TYPE(f.budget_j) = 'INTEGER' + AND CAST(f.timeout_j AS SIGNED) = CAST(f.budget_j AS SIGNED)) IS NOT TRUE + +UNION ALL +-- When a per-chain map is bound, each chain's original T equals its entry. +SELECT CONVERT('B2b chain budget differs from its expected entry' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(f.run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(COALESCE(f.occ, 'NULL'), ' budget=', COALESCE(CAST(f.budget_j AS CHAR CHARACTER SET utf8mb4) COLLATE utf8mb4_bin, 'NULL'), ' expected=', + COALESCE(CAST(JSON_EXTRACT(@expected_budgets, CONCAT('$."', f.run_id COLLATE utf8mb4_bin, '|', f.occ, '"')) AS CHAR CHARACTER SET utf8mb4) COLLATE utf8mb4_bin, 'missing')) USING utf8mb4) COLLATE utf8mb4_bin +FROM firsts f +WHERE @expected_budgets IS NOT NULL + AND (CAST(JSON_EXTRACT(@expected_budgets, CONCAT('$."', f.run_id COLLATE utf8mb4_bin, '|', f.occ, '"')) AS SIGNED) + = CAST(f.budget_j AS SIGNED)) IS NOT TRUE + +UNION ALL +SELECT CONVERT('B2c expected budget names no present chain' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(NULL USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(k.chain USING utf8mb4) COLLATE utf8mb4_bin +FROM JSON_TABLE(COALESCE(JSON_KEYS(@expected_budgets), JSON_ARRAY()), '$[*]' COLUMNS (chain VARCHAR(300) PATH '$')) k +WHERE @expected_budgets IS NOT NULL + AND NOT EXISTS (SELECT 1 FROM firsts f WHERE CONCAT(f.run_id COLLATE utf8mb4_bin, '|', f.occ) = CONVERT(k.chain USING utf8mb4) COLLATE utf8mb4_bin) + +UNION ALL +-- Every open of one run/occurrence keeps the first open's budget, key, +-- fingerprint and recorded anchor string. +SELECT CONVERT('B3 reopen changed its chain identity, budget or anchor' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(o.run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(COALESCE(o.wait_id, 'NULL'), ' vs first ', COALESCE(a.first_wait, 'NULL'), + ' budget ', COALESCE(CAST(o.budget_j AS CHAR CHARACTER SET utf8mb4) COLLATE utf8mb4_bin, 'NULL'), '/', COALESCE(CAST(a.budget_j AS CHAR CHARACTER SET utf8mb4) COLLATE utf8mb4_bin, 'NULL'), + ' anchor ', COALESCE(o.anchor_s, 'NULL'), '/', COALESCE(a.anchor_s, 'NULL')) USING utf8mb4) COLLATE utf8mb4_bin +FROM opens o JOIN anchors a ON a.run_id = o.run_id AND a.occ <=> o.occ +WHERE (CAST(o.budget_j AS SIGNED) = CAST(a.budget_j AS SIGNED) AND o.ckey <=> a.ckey AND o.fp <=> a.fp + AND o.anchor_s = a.anchor_s) IS NOT TRUE + +UNION ALL +-- B8: exact one-to-one identities, so no join below can hide a duplicate. +SELECT CONVERT('B8a duplicate physical open identity' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(o.run_id USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(CONCAT(COALESCE(o.wait_id, 'NULL'), ' opens=', COUNT(*)) USING utf8mb4) COLLATE utf8mb4_bin +FROM opens o GROUP BY o.run_id, o.wait_id HAVING COUNT(*) <> 1 + +UNION ALL +SELECT CONVERT('B8b open without exactly one condition-timeout schedule' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(o.run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(COALESCE(o.wait_id, 'NULL'), ' schedules=', + (SELECT COUNT(*) FROM sched s WHERE s.run_id = o.run_id AND s.wait_id <=> o.wait_id)) USING utf8mb4) COLLATE utf8mb4_bin +FROM opens o +WHERE ((SELECT COUNT(*) FROM sched s WHERE s.run_id = o.run_id AND s.wait_id <=> o.wait_id) = 1) IS NOT TRUE + +UNION ALL +SELECT CONVERT('B8c orphan condition-timeout schedule' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(s.run_id USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(CONCAT(COALESCE(s.wait_id, 'NULL'), ' timer=', COALESCE(s.timer_id, 'NULL')) USING utf8mb4) COLLATE utf8mb4_bin +FROM sched s +WHERE NOT EXISTS (SELECT 1 FROM opens o WHERE o.run_id = s.run_id AND o.wait_id <=> s.wait_id) + +UNION ALL +SELECT CONVERT('B8d schedule identity not a string, timer shared or without its row' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(s.run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(COALESCE(s.wait_id, 'NULL'), ' timer=', COALESCE(s.timer_id, 'NULL')) USING utf8mb4) COLLATE utf8mb4_bin +FROM sched s +WHERE (s.wait_t = 'STRING' AND s.timer_t = 'STRING' AND s.fire_t = 'STRING' AND s.policy_t = 'STRING' AND s.anchor_t = 'STRING' + AND s.wait_id <> '' AND s.timer_id <> '' + AND (SELECT COUNT(*) FROM sched d WHERE d.timer_id = s.timer_id) = 1 + AND (SELECT COUNT(*) FROM workflow_run_timers t WHERE t.id COLLATE utf8mb4_bin = s.timer_id AND t.workflow_run_id = s.run_id) = 1) IS NOT TRUE + +UNION ALL +-- The first open's recorded anchor is exactly its own timer row's fire_at +-- (Native writes microsecond RFC 3339 with Z). +SELECT CONVERT('B4 first open anchor is not its own timer fire_at' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(a.run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(COALESCE(a.first_wait, 'NULL'), ' timer=', COALESCE(a.timer_id, 'missing'), ' fire_at=', + COALESCE(CONVERT(DATE_FORMAT(a.a_at, '%Y-%m-%dT%H:%i:%s.%fZ') USING utf8mb4) COLLATE utf8mb4_bin, 'NULL'), ' recorded=', COALESCE(a.anchor_s, 'NULL')) USING utf8mb4) COLLATE utf8mb4_bin +FROM anchors a +WHERE (CONVERT(DATE_FORMAT(a.a_at, '%Y-%m-%dT%H:%i:%s.%fZ') USING utf8mb4) COLLATE utf8mb4_bin = a.anchor_s) IS NOT TRUE + +UNION ALL +-- Every open, the first included, scheduled its timer at exactly A. +SELECT CONVERT('B5 chain timer not scheduled exactly at the first anchor' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(o.run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(COALESCE(o.wait_id, 'NULL'), ' timer=', COALESCE(s.timer_id, 'missing'), ' scheduled=', COALESCE(s.fire_s, 'NULL'), + ' policy=', COALESCE(s.policy, 'NULL'), ' row_fire_at=', + COALESCE(CONVERT(DATE_FORMAT(t.fire_at, '%Y-%m-%dT%H:%i:%s.%fZ') USING utf8mb4) COLLATE utf8mb4_bin, 'NULL'), ' A=', COALESCE(a.anchor_s, 'NULL')) USING utf8mb4) COLLATE utf8mb4_bin +FROM opens o +JOIN anchors a ON a.run_id = o.run_id AND a.occ <=> o.occ +LEFT JOIN sched s ON s.run_id = o.run_id AND s.wait_id = o.wait_id +LEFT JOIN workflow_run_timers t ON t.id COLLATE utf8mb4_bin = s.timer_id AND t.workflow_run_id = o.run_id +WHERE (s.policy = 'occurrence' AND s.anchor_s = a.anchor_s AND s.fire_s = a.anchor_s AND t.fire_at = a.a_at) IS NOT TRUE + +UNION ALL +-- The waits projection carries A as every chain wait's deadline. +SELECT CONVERT('B6 projected wait deadline is not the first anchor' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(o.run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(COALESCE(o.wait_id, 'NULL'), ' deadline_at=', COALESCE(CONVERT(DATE_FORMAT(w.deadline_at, '%Y-%m-%dT%H:%i:%s.%fZ') USING utf8mb4) COLLATE utf8mb4_bin, 'missing'), + ' A=', COALESCE(a.anchor_s, 'NULL')) USING utf8mb4) COLLATE utf8mb4_bin +FROM opens o +JOIN anchors a ON a.run_id = o.run_id AND a.occ <=> o.occ +LEFT JOIN workflow_run_waits w ON w.workflow_run_id = o.run_id AND w.wait_id COLLATE utf8mb4_bin = o.wait_id AND w.kind = 'condition' +WHERE (w.deadline_at = a.a_at) IS NOT TRUE + +UNION ALL +-- A fired chain timer is never early and fires within the selected bound. +SELECT CONVERT('B7 chain timer fired early or beyond the dispatch bound' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(o.run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(t.id, ' fire_at=', CONVERT(DATE_FORMAT(t.fire_at, '%Y-%m-%dT%H:%i:%s.%fZ') USING utf8mb4) COLLATE utf8mb4_bin, + ' fired_at=', COALESCE(CONVERT(DATE_FORMAT(t.fired_at, '%Y-%m-%dT%H:%i:%s.%fZ') USING utf8mb4) COLLATE utf8mb4_bin, 'NULL')) USING utf8mb4) COLLATE utf8mb4_bin +FROM opens o +JOIN sched s ON s.run_id = o.run_id AND s.wait_id = o.wait_id +JOIN workflow_run_timers t ON t.id COLLATE utf8mb4_bin = s.timer_id AND t.workflow_run_id = o.run_id +WHERE t.status = 'fired' + AND (t.fired_at >= t.fire_at + AND TIMESTAMPDIFF(MICROSECOND, t.fire_at, t.fired_at) <= @max_dispatch_seconds * 1000000) IS NOT TRUE + +UNION ALL +-- The original A1, A2, A3, A4, A5, A6 and A0 predicates, unchanged in meaning. +-- Server#1098 changed no projection status, timer settlement or task rule. +SELECT CONVERT('A1 open condition wait' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(w.workflow_run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(w.wait_id, ' status=', w.status, ' source_status=', COALESCE(w.source_status, 'NULL')) USING utf8mb4) COLLATE utf8mb4_bin +FROM workflow_run_waits w +WHERE FIND_IN_SET(w.workflow_run_id, CONVERT(@run_ids USING utf8mb4) COLLATE utf8mb4_bin) AND w.kind = 'condition' AND w.status = 'open' + +UNION ALL +SELECT CONVERT('A2 condition wait not settled to a native terminal pair' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(w.workflow_run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(w.wait_id, ' ', w.status, '/', COALESCE(w.source_status, 'NULL'), ' resolved_at=', COALESCE(w.resolved_at, 'NULL')) USING utf8mb4) COLLATE utf8mb4_bin +FROM workflow_run_waits w +WHERE FIND_IN_SET(w.workflow_run_id, CONVERT(@run_ids USING utf8mb4) COLLATE utf8mb4_bin) AND w.kind = 'condition' + AND NOT (w.resolved_at IS NOT NULL AND ( + (w.status = 'resolved' AND w.source_status = 'satisfied') + OR (w.status = 'cancelled' AND w.source_status IN ('cancelled', 'terminated', 'continued', 'closed')))) + +UNION ALL +SELECT CONVERT('A3 condition timeout timer not fired or cancelled' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(t.workflow_run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(t.id, ' seq=', t.sequence, ' status=', t.status, ' fire_at=', t.fire_at) USING utf8mb4) COLLATE utf8mb4_bin +FROM workflow_run_timers t +WHERE FIND_IN_SET(t.workflow_run_id, CONVERT(@run_ids USING utf8mb4) COLLATE utf8mb4_bin) AND t.status NOT IN ('fired', 'cancelled') + AND EXISTS (SELECT 1 FROM workflow_run_waits w WHERE w.workflow_run_id = t.workflow_run_id + AND w.kind = 'condition' AND w.sequence = t.sequence) + +UNION ALL +SELECT CONVERT('A4 unexpected timer entry row' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(e.workflow_run_id USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(CONCAT(e.id, ' status=', COALESCE(e.status, 'NULL')) USING utf8mb4) COLLATE utf8mb4_bin +FROM workflow_run_timer_entries e +WHERE FIND_IN_SET(e.workflow_run_id, CONVERT(@run_ids USING utf8mb4) COLLATE utf8mb4_bin) + +UNION ALL +SELECT CONVERT('A5 runnable condition timeout task after close' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(k.workflow_run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(k.id, ' status=', k.status, ' available_at=', COALESCE(k.available_at, 'NULL')) USING utf8mb4) COLLATE utf8mb4_bin +FROM workflow_tasks k +WHERE FIND_IN_SET(k.workflow_run_id, CONVERT(@run_ids USING utf8mb4) COLLATE utf8mb4_bin) AND k.task_type = 'timer' AND k.status IN ('ready', 'leased') + AND JSON_EXTRACT(k.payload, '$.condition_wait_id') IS NOT NULL + +UNION ALL +SELECT CONVERT('A6 overlapping condition waits for one key' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(a.workflow_run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(COALESCE(a.target_name, 'NULL'), ': ', a.wait_id, ' and ', b.wait_id) USING utf8mb4) COLLATE utf8mb4_bin +FROM workflow_run_waits a +JOIN workflow_run_waits b ON b.workflow_run_id = a.workflow_run_id AND b.kind = 'condition' + AND b.target_name <=> a.target_name AND b.wait_id > a.wait_id +WHERE FIND_IN_SET(a.workflow_run_id, CONVERT(@run_ids USING utf8mb4) COLLATE utf8mb4_bin) AND a.kind = 'condition' + AND a.opened_at < COALESCE(b.resolved_at, '9999-12-31') AND b.opened_at < COALESCE(a.resolved_at, '9999-12-31') + +UNION ALL +SELECT CONVERT('A6 condition rows differ from ConditionWaitOpened events' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(r.id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT('rows=', (SELECT COUNT(*) FROM workflow_run_waits w WHERE w.workflow_run_id = r.id AND w.kind = 'condition'), + ' events=', (SELECT COUNT(*) FROM workflow_history_events h WHERE h.workflow_run_id = r.id AND h.event_type = 'ConditionWaitOpened')) USING utf8mb4) COLLATE utf8mb4_bin +FROM workflow_runs r +WHERE FIND_IN_SET(r.id, CONVERT(@run_ids USING utf8mb4) COLLATE utf8mb4_bin) + AND (SELECT COUNT(*) FROM workflow_run_waits w WHERE w.workflow_run_id = r.id AND w.kind = 'condition') + <> (SELECT COUNT(*) FROM workflow_history_events h WHERE h.workflow_run_id = r.id AND h.event_type = 'ConditionWaitOpened') + +UNION ALL +SELECT CONVERT('A0 run missing or still open' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(ids.id USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(COALESCE(r.status, 'missing') USING utf8mb4) COLLATE utf8mb4_bin +FROM JSON_TABLE(CONCAT('["', REPLACE(CONVERT(@run_ids USING utf8mb4) COLLATE utf8mb4_bin, ',', '","'), '"]'), '$[*]' COLUMNS (id VARCHAR(26) PATH '$')) ids +LEFT JOIN workflow_runs r ON r.id COLLATE utf8mb4_bin = CONVERT(ids.id USING utf8mb4) COLLATE utf8mb4_bin +WHERE r.id IS NULL OR r.status NOT IN ('completed', 'failed', 'cancelled', 'terminated'); diff --git a/scripts/conformance/php-sdk-condition-wait-export.php b/scripts/conformance/php-sdk-condition-wait-export.php new file mode 100644 index 000000000..cf9a38a9a --- /dev/null +++ b/scripts/conformance/php-sdk-condition-wait-export.php @@ -0,0 +1,64 @@ + $workflowId) { + $export = $client->exportWorkflowHistory($workflowId, $runId); + $handle = fopen($root.'/php-condition-wait-export-'.$runId.'.json', 'xb'); + if ($handle === false) { + throw new RuntimeException('condition_export_output_exists'); + } + $bytes = json_encode($export, JSON_THROW_ON_ERROR | JSON_PRETTY_PRINT).PHP_EOL; + if (fwrite($handle, $bytes) !== strlen($bytes)) { + throw new RuntimeException('condition_export_short_write'); + } + fclose($handle); + } + echo json_encode(['exports' => count($runs), 'outcome' => 'pass'], JSON_THROW_ON_ERROR).PHP_EOL; +} catch (Throwable $error) { + $message = 'Condition export failed: '.get_class($error); + if ($error instanceof ServerException) { + $message .= ' status='.$error->status; + if (preg_match('/\A[a-z0-9_]{1,80}\z/D', (string) $error->reason) === 1) { + $message .= ' reason='.$error->reason; + } + } elseif (preg_match('/\Acondition_export_[a-z_]{1,60}\z/D', $error->getMessage()) === 1) { + $message .= ' reason='.$error->getMessage(); + } + fwrite(STDERR, $message.PHP_EOL); + exit(1); +} diff --git a/scripts/conformance/php-sdk-condition-wait-input.py b/scripts/conformance/php-sdk-condition-wait-input.py new file mode 100644 index 000000000..c8f8ee3d6 --- /dev/null +++ b/scripts/conformance/php-sdk-condition-wait-input.py @@ -0,0 +1,159 @@ +#!/usr/bin/env python3 +"""Bind the installed C1-C5 runs and raw exports to the reviewed read-only SQL.""" +import hashlib +import json +from pathlib import Path +import re +import sys + +CASES = ('C1', 'C2', 'C3-terminate', 'C3-cancel', 'C3-continue', 'C4', 'C5') +SQL_SHA256 = '1f828778b952b931fd99acccc05af415aab1e279f2713cc164f737a99337b21c' +RUN_ID = re.compile(r'[0-9A-HJKMNP-TV-Z]{26}') + + +def require(condition, reason): + if not condition: + raise ValueError(reason) + + +def report_runs(report): + require(isinstance(report, dict), 'condition_report_shape') + require(isinstance(report.get('namespace'), str) + and re.fullmatch(r'php-cwn-[a-f0-9]{12}', report['namespace']) is not None, + 'condition_namespace') + cases = report.get('cases') + require(isinstance(cases, dict) and set(cases) == set(CASES), 'condition_cases_missing') + rows = [] + for case in CASES: + entry = cases[case] + require(isinstance(entry, dict) and entry.get('evidence'), 'condition_evidence_missing') + require(isinstance(entry.get('command_contract'), dict) + and entry['command_contract'].get('source') == 'durable_history', + 'condition_contract_not_durable') + runs = entry.get('runs') + require(isinstance(runs, list) and len(runs) == (2 if case in ('C3-continue', 'C5') else 1), + 'condition_run_count') + for run in runs: + require(isinstance(run, dict), 'condition_run_shape') + require(isinstance(run.get('run_id'), str) and RUN_ID.fullmatch(run['run_id']) is not None, + 'condition_run_id') + require(isinstance(run.get('workflow_id'), str) + and re.fullmatch(r'c[1-5]-[a-z0-9-]+', run['workflow_id']) is not None, + 'condition_workflow_id') + require(isinstance(run.get('events'), list) and run['events'] + and all(isinstance(event, str) and event for event in run['events']), + 'condition_events_missing') + rows.append((case, run)) + ids = [run['run_id'] for _, run in rows] + require(len(ids) == 9 and len(set(ids)) == 9 and report.get('run_ids') == ids, + 'condition_run_inventory') + require(report.get('c1_run_ids') == [rows[0][1]['run_id']] and report.get('c1_timeout') == 20, + 'condition_first_case_identity') + return rows + + +def extract_report(raw): + """Keep the published CLI's progress separate from its final JSON object.""" + lines = raw.splitlines(keepends=True) + progress = 0 + for line in lines: + if not line.startswith(b'ok: '): + break + progress += 1 + require(progress > 0, 'condition_cli_progress_missing') + report = json.loads(b''.join(lines[progress:])) + report_runs(report) + return report + + +def validate_exports(report, exports): + rows = report_runs(report) + for case, run in rows: + export = exports.get(run['run_id']) + require(isinstance(export, dict) and export.get('schema') == 'durable-workflow.v2.history-export' + and export.get('schema_version') == 2 and export.get('history_complete') is True, + 'condition_export_incomplete') + workflow = export.get('workflow', {}) + require(isinstance(workflow, dict) + and workflow.get('run_id') == run['run_id'] and workflow.get('instance_id') == run['workflow_id'] + and workflow.get('source_runtime') == 'standalone_server' + and workflow.get('source_process_class') == 'server_http_node' + and 'migration_projection' not in export, 'condition_export_identity') + events = export.get('history_events') + require(isinstance(events, list) and len(events) >= 2, 'condition_export_events') + identities = set() + for sequence, event in enumerate(events, 1): + require(isinstance(event, dict) and type(event.get('sequence')) is int + and event['sequence'] == sequence and isinstance(event.get('id'), str) + and event['id'] and event['id'] not in identities, 'condition_export_prefix') + identities.add(event['id']) + require([event.get('type') for event in events] == run['events'], 'condition_export_history_mismatch') + pair = report['cases'][case]['runs'] + if len(pair) == 2 and run['run_id'] == pair[1]['run_id']: + parent = pair[0] + started = events[0].get('payload', {}) + parent_events = exports[parent['run_id']]['history_events'] + links = [event.get('payload', {}) for event in parent_events + if event.get('type') == 'WorkflowContinuedAsNew'] + require(events[0]['type'] == 'WorkflowStarted' and isinstance(started, dict) + and events[0].get('workflow_task_id') is None + and events[0].get('workflow_command_id') is None + and parent['workflow_id'] == run['workflow_id'] + and started.get('continued_from_run_id') == parent['run_id'] + and started.get('workflow_run_id') == run['run_id'] + and started.get('workflow_instance_id') == run['workflow_id'] + and isinstance(started.get('workflow_link_id'), str) and started['workflow_link_id'] + and len(links) == 1 and isinstance(links[0], dict) + and links[0].get('continued_to_run_id') == run['run_id'] + and links[0].get('workflow_link_id') == started['workflow_link_id'], + 'condition_export_continuation_authority') + else: + require(events[0]['type'] == 'StartAccepted' and events[1]['type'] == 'WorkflowStarted' + and events[0].get('workflow_task_id') is None and events[1].get('workflow_task_id') is None + and isinstance(events[0].get('workflow_command_id'), str) + and events[0]['workflow_command_id'] + and events[0]['workflow_command_id'] == events[1].get('workflow_command_id'), + 'condition_export_start_authority') + if case == 'C2': + kinds = [event['type'] for event in events] + required = ('ActivityScheduled', 'ActivityCompleted', 'ConditionWaitOpened', 'WorkflowCompleted') + require(all(kind in kinds for kind in required), 'condition_c2_activity_missing') + scheduled, completed, opened = (kinds.index(kind) for kind in required[:3]) + noise = [index for index, event in enumerate(events) if event['type'] == 'SignalReceived' + and isinstance(event.get('payload'), dict) and event['payload'].get('signal_name') == 'noise'] + require(any(scheduled < index < completed < opened for index in noise), + 'condition_c2_signal_not_during_activity') + return rows + + +def sql_input(report, exports, sql): + rows = validate_exports(report, exports) + require(hashlib.sha256(sql).hexdigest() == SQL_SHA256, 'condition_sql_identity') + ids = ','.join(run['run_id'] for _, run in rows) + return ("SET SESSION time_zone = '+00:00';\nSET @run_ids = '" + ids + "';\n" + "SET @expected_namespace = '" + report['namespace'] + "';\n" + "SET @max_dispatch_seconds = 2;\nSET @expected_budgets = NULL;\n" + "START TRANSACTION READ ONLY;\n" + "SELECT 'R0_run_inventory', COUNT(*) FROM workflow_runs " + "WHERE FIND_IN_SET(id COLLATE utf8mb4_bin, CONVERT(@run_ids USING utf8mb4) COLLATE utf8mb4_bin) " + "AND namespace COLLATE utf8mb4_bin = CONVERT(@expected_namespace USING utf8mb4) COLLATE utf8mb4_bin " + "HAVING COUNT(*) <> 9;\n" + sql.decode('utf-8') + "\nROLLBACK;\n") + + +def main(): + if len(sys.argv) == 3 and sys.argv[1] == '--extract-report': + root = Path(sys.argv[2]) + report = extract_report((root / 'php-condition-waits.stdout.log').read_bytes()) + sys.stdout.write(json.dumps(report, indent=2) + '\n') + return + root = Path(sys.argv[1]) + report = json.loads((root / 'php-condition-waits.json').read_bytes()) + rows = report_runs(report) + exports = {run['run_id']: json.loads((root / ('php-condition-wait-export-' + run['run_id'] + '.json')).read_bytes()) + for _, run in rows} + sql = Path(__file__).with_name('condition-wait-anchored-check.sql').read_bytes() + sys.stdout.write(sql_input(report, exports, sql)) + + +if __name__ == '__main__': + main() diff --git a/scripts/conformance/php-sdk-replay-host-published-artifacts.sh b/scripts/conformance/php-sdk-replay-host-published-artifacts.sh index 55cb4e56b..771ac6ed0 100755 --- a/scripts/conformance/php-sdk-replay-host-published-artifacts.sh +++ b/scripts/conformance/php-sdk-replay-host-published-artifacts.sh @@ -10,11 +10,13 @@ fi server_image="${DW_SERVER_IMAGE:-}" runner_image="${DW_REPLAY_PHP_RUNNER_IMAGE:-}" mysql_image="${DW_WORKFLOW_LIFECYCLE_MYSQL_IMAGE:-}" +condition_waits="${DW_PHP_SDK_CONFORMANCE_CONDITION_WAITS:-0}" if [[ ! "${DW_SERVER_VERSION:-}" =~ ^3\.[0-9]+\.[0-9]+$ \ || ! "${DW_PHP_SDK_VERSION:-}" =~ ^[0-9]+\.[0-9]+\.[0-9]+$ \ || ! "$server_image" =~ ^(docker.io/)?durableworkflow/server@sha256:[0-9a-f]{64}$ \ || ! "$runner_image" =~ ^ghcr.io/durable-workflow/sample-app-devcontainer@sha256:[0-9a-f]{64}$ \ - || ! "$mysql_image" =~ ^(docker.io/library/)?mysql@sha256:[0-9a-f]{64}$ ]]; then + || ! "$mysql_image" =~ ^(docker.io/library/)?mysql@sha256:[0-9a-f]{64}$ \ + || ! "$condition_waits" =~ ^[01]$ ]]; then echo 'PHP replay requires exact Native/SDK versions and digest-pinned Server, MySQL and published tooling images.' >&2 exit 2 fi @@ -134,7 +136,8 @@ docker create --name "$server" --label "$label" --network "$network" \ -e DB_CONNECTION=mysql -e DB_HOST="$mysql" -e DB_PORT=3306 \ -e DB_DATABASE=durable_workflow -e DB_USERNAME=workflow -e DB_PASSWORD=workflow \ -e DW_BIND_ADDRESS=0.0.0.0:8080 -e DW_AUTH_DRIVER=token \ - -e DW_AUTH_TOKEN=php-replay-control -e DW_WORKER_TOKEN=php-replay-worker \ + -e DW_AUTH_TOKEN=claimed-task-admin -e DW_OPERATOR_TOKEN=claimed-task-client \ + -e DW_WORKER_TOKEN=claimed-task-worker \ "$server_image" >"$result_dir/php-replay-server-id.txt" grep -Ex '[0-9a-f]{64}' "$result_dir/php-replay-server-id.txt" >/dev/null creating='' @@ -150,9 +153,10 @@ docker create --name "$probe" --label "$label" --network "$network" \ -e DW_SERVER_VERSION -e DW_SERVER_IMAGE -e DW_PHP_SDK_VERSION \ -e DW_PHP_SDK_CONFORMANCE_SERVER_URL="http://$server:8080" \ -e DW_PHP_SDK_CONFORMANCE_NAMESPACE=php-replay-conformance \ - -e DW_PHP_SDK_CONFORMANCE_TOKEN=php-replay-control \ - -e DW_PHP_SDK_CONFORMANCE_WORKER_TOKEN=php-replay-worker \ + -e DW_PHP_SDK_CONFORMANCE_TOKEN=claimed-task-admin \ + -e DW_PHP_SDK_CONFORMANCE_WORKER_TOKEN=claimed-task-worker \ -e DW_PHP_SDK_CONFORMANCE_REPLAY_MATRIX=1 \ + -e DW_PHP_SDK_CONFORMANCE_CONDITION_WAITS="$condition_waits" \ --entrypoint bash "$runner_image" -euc ' mkdir -p -m 0700 "$HOME/bin" # The image PHP has a low-port file capability, which cannot execute with @@ -182,6 +186,17 @@ docker create --name "$probe" --label "$label" --network "$network" \ --data "{\"name\":\"$DW_PHP_SDK_CONFORMANCE_NAMESPACE\",\"retention_days\":7}" \ "$DW_PHP_SDK_CONFORMANCE_SERVER_URL/api/namespaces" > /result/php-replay-namespace.json bash /runner/php-sdk-published-artifacts.sh --result-dir /result + if [[ "$DW_PHP_SDK_CONFORMANCE_CONDITION_WAITS" == 1 ]]; then + # Run the qualifier from the normal installed package, with its own + # synthetic namespace and worker. The original replay matrix stays intact. + php -l /runner/php-sdk-condition-wait-export.php > /result/php-condition-export-lint.txt + php /result/php-sdk-project/vendor/durable-workflow/sdk/scripts/qualify-condition-wait-native.php \ + /result/php-sdk-project/vendor/autoload.php "$DW_PHP_SDK_CONFORMANCE_SERVER_URL" \ + > /result/php-condition-waits.stdout.log 2> /result/php-condition-waits.log + python3 /runner/php-sdk-condition-wait-input.py --extract-report /result > /result/php-condition-waits.json + php /runner/php-sdk-condition-wait-export.php > /result/php-condition-wait-exports.json + python3 /runner/php-sdk-condition-wait-input.py /result > /result/php-condition-wait-input.sql + fi python3 /runner/php-sdk-replay-result.py /result ' >"$result_dir/php-replay-probe-id.txt" grep -Ex '[0-9a-f]{64}' "$result_dir/php-replay-probe-id.txt" >/dev/null @@ -189,3 +204,15 @@ creating='' docker start --attach "$probe" >"$result_dir/php-replay-shard.log" 2>&1 test "$(query container inspect --format '{{.State.Status}}' "$probe")" = exited test "$(query container inspect --format '{{.State.ExitCode}}' "$probe")" = 0 +if [[ "$condition_waits" == 1 ]]; then + # The reviewed checker is SELECT-only, and the session is read-only too. + timeout 60 docker exec -i -e MYSQL_PWD=workflow "$mysql" mysql \ + --default-character-set=utf8mb4 --batch --raw --skip-column-names -uworkflow durable_workflow \ + < "$result_dir/php-condition-wait-input.sql" \ + > "$result_dir/php-condition-wait-sql.tsv" 2> "$result_dir/php-condition-wait-sql.err" + test ! -s "$result_dir/php-condition-wait-sql.err" + test ! -s "$result_dir/php-condition-wait-sql.tsv" + sha256sum "$result_dir/php-condition-wait-input.sql" > "$result_dir/php-condition-wait-sql-sha256.txt" + printf '{"outcome":"pass","violations":0,"readonly_transaction":true}\n' \ + > "$result_dir/php-condition-wait-sql-check.json" +fi diff --git a/scripts/conformance/server-parity/changes.mjs b/scripts/conformance/server-parity/changes.mjs index fdd59540c..cf4b3da87 100644 --- a/scripts/conformance/server-parity/changes.mjs +++ b/scripts/conformance/server-parity/changes.mjs @@ -28,10 +28,15 @@ export function pushPathPatterns(workflowText) { let inPaths = false; for (const line of lines.slice(push + 1)) { if (/^ {0,3}\S/.test(line)) break; + if (/^ {4,}#/.test(line)) continue; if (line === ' paths:') { inPaths = true; } else if (inPaths && line.startsWith(' - ')) { - patterns.push(line.slice(' - '.length).trim()); + // YAML requires quoting a leading ! so it is not read as a tag. + const entry = line.slice(' - '.length).trim(); + const quoted = /^'(![^']+)'$/.exec(entry); + if (!quoted && /^['"!]/.test(entry)) throw new Error(`Unsupported path filter entry: ${entry}`); + patterns.push(quoted ? quoted[1] : entry); } else if (inPaths) { break; } @@ -66,8 +71,12 @@ export function nulSeparatedPaths(bytes) { } export function matchingFiles(patterns, files) { - const expressions = patterns.map(patternRegExp); - return files.filter(file => expressions.some(expression => expression.test(file))); + // GitHub evaluates paths in order: the last matching pattern decides. + const rules = patterns.map(pattern => pattern.startsWith('!') + ? {include: false, expression: patternRegExp(pattern.slice(1))} + : {include: true, expression: patternRegExp(pattern)}); + if (rules.length > 0 && !rules[0].include) throw new Error('A path filter cannot start with a negated pattern'); + return files.filter(file => rules.reduce((selected, rule) => rule.expression.test(file) ? rule.include : selected, false)); } function classify({event, headSha, mergeBaseRef, mergeBaseSha, mergeHeadSha, workflowPath}) { diff --git a/tests/Unit/ServerParityAggregateTest.mjs b/tests/Unit/ServerParityAggregateTest.mjs index 2f39690da..718cb16d7 100644 --- a/tests/Unit/ServerParityAggregateTest.mjs +++ b/tests/Unit/ServerParityAggregateTest.mjs @@ -71,6 +71,34 @@ test('path patterns follow GitHub filter semantics for the forms in use', () => assert.deepEqual(matchingFiles(['rust/**', 'docs/x.md'], ['README.md', 'rust/Cargo.toml', 'docs/x.md']), ['rust/Cargo.toml', 'docs/x.md']); }); +test('only the publication subject is excluded from CI-helper fixture inputs', () => { + const patterns = pushPathPatterns(workflowText); + const subject = 'scripts/ci/native-qualification-subject.json'; + assert.deepEqual(matchingFiles(patterns, [subject]), []); + // These executable helpers remain selected, including the isolation check + // that the changes job now executes before the expensive fixture fan-out. + const selected = ['rust/src/runtime/store.rs', 'Cargo.lock', '.github/workflows/server-parity.yml', + 'scripts/ci/check-docker-workflow-isolation.py', 'scripts/ci/native_alias_evidence.py', + 'scripts/ci/native_qualification.py', 'scripts/ci/new-helper.sh', + 'scripts/ci/nested/native-qualification-subject.json', + 'scripts/conformance/server-parity/native-image-corpus.sh']; + for (const file of selected) { + assert.deepEqual(matchingFiles(patterns, [file]), [file], `${file} must re-run the fixtures`); + assert.deepEqual(matchingFiles(patterns, [subject, file]), [file], `a subject change must not hide ${file}`); + } +}); + +test('negated path filters preserve GitHub ordering and refuse ambiguous YAML entries', () => { + assert.deepEqual(matchingFiles(['a/**', '!a/b.py', 'a/b*'], ['a/b.py', 'a/c.py', 'x.py']), ['a/b.py', 'a/c.py']); + assert.deepEqual(matchingFiles(['a/**', '!a/*.py'], ['a/b.py', 'a/b/c.py']), ['a/b/c.py']); + assert.throws(() => matchingFiles(['!a/**', 'b/**'], ['b/c']), /cannot start with a negated pattern/); + assert.throws(() => patternRegExp('!a/**'), /Unsupported path filter pattern/); + const listed = entries => pushPathPatterns(`on:\n push:\n paths:\n${entries}\njobs:\n`); + assert.deepEqual(listed(" - a/**\n # Ordered exclusion\n - '!a/b.py'"), ['a/**', '!a/b.py']); + assert.throws(() => listed(' - a/**\n - !a/b.py'), /Unsupported path filter entry/); + assert.throws(() => listed(" - a/**\n - '!a/b.py"), /Unsupported path filter entry/); +}); + test('literal native inputs and parity imports are covered by the shared input list', () => { const sourceFiles = (directory, suffix) => readdirSync(directory, {withFileTypes: true}).flatMap(entry => { const path = join(directory, entry.name); @@ -331,6 +359,12 @@ function classify(root, env) { for (const [label, change, expected] of [ ['a documentation-only change skips the fixtures', ({write}) => write('docs/note.md', 'x\n'), 'run=false'], + ['a publication-subject-only change skips the fixtures', ({write}) => + write('scripts/ci/native-qualification-subject.json', '{}\n'), 'run=false'], + ['a publication subject cannot hide a Rust change', ({write}) => { + write('scripts/ci/native-qualification-subject.json', '{}\n'); + write('rust/src/lib.rs', 'x\n'); + }, 'run=true'], ['a Rust change runs the fixtures', ({write}) => write('rust/src/lib.rs', 'x\n'), 'run=true'], ['a configuration-audit contract edit runs the fixtures', ({write}) => write('config/dw-contract.php', " ['env' => 'FIXTURE_OPTION']];\n"), 'run=true'], @@ -471,6 +505,8 @@ test('the workflow supplies the merge-group identity to the classifier', () => { for (const [label, change, expected] of [ ['PHP-only changes skip the fixtures', ({write}) => write('app/Controller.php', ' write('docs/note.md', 'x\n'), 'run=false'], + ['a publication-subject-only change skips the fixtures', ({write}) => + write('scripts/ci/native-qualification-subject.json', '{}\n'), 'run=false'], ['native changes run the fixtures', ({write}) => write('rust/src/lib.rs', 'x\n'), 'run=true'], ['parent Cargo configuration runs the fixtures', ({write}) => write('.cargo/config.toml', '[build]\n'), 'run=true'], ['parent Cargo manifests run the fixtures', ({write}) => write('Cargo.toml', '[workspace]\n'), 'run=true'], diff --git a/tests/Unit/Support/php_condition_wait_test.py b/tests/Unit/Support/php_condition_wait_test.py new file mode 100644 index 000000000..d04e6dee8 --- /dev/null +++ b/tests/Unit/Support/php_condition_wait_test.py @@ -0,0 +1,154 @@ +import importlib.util +import json +from pathlib import Path +import unittest + +RUNNERS = Path(__file__).resolve().parents[3] / 'scripts/conformance' +spec = importlib.util.spec_from_file_location('condition_input', RUNNERS / 'php-sdk-condition-wait-input.py') +gate = importlib.util.module_from_spec(spec) +spec.loader.exec_module(gate) + + +class InstalledConditionWaitInputTest(unittest.TestCase): + def fixture(self): + report = {'namespace': 'php-cwn-0123456789ab', 'cases': {}, 'run_ids': [], 'c1_timeout': 20} + exports = {} + number = 0 + for case in gate.CASES: + rows = [] + for position in range(2 if case in ('C3-continue', 'C5') else 1): + number += 1 + run_id = str(number).zfill(26) + workflow_id = case.lower() + '-0123456789ab' + kinds = ['StartAccepted', 'WorkflowStarted', 'ConditionWaitOpened', 'WorkflowCompleted'] + if case == 'C2': + kinds = ['StartAccepted', 'WorkflowStarted', 'ActivityScheduled', 'ActivityStarted', + 'SignalReceived', 'ActivityCompleted', 'ConditionWaitOpened', 'WorkflowCompleted'] + if position == 1: + kinds = ['WorkflowStarted', 'WorkflowCompleted'] + events = [{'id': run_id + '-event-' + str(index), 'sequence': index, 'type': kind, + 'workflow_command_id': 'start-authority' if index <= 2 else None, + 'payload': {'signal_name': 'noise'} if kind == 'SignalReceived' else {}} + for index, kind in enumerate(kinds, 1)] + if position == 1: + parent = rows[0] + link = 'continue-link-' + run_id + events[0]['workflow_command_id'] = None + events[0]['payload'] = {'continued_from_run_id': parent['run_id'], 'workflow_run_id': run_id, + 'workflow_instance_id': workflow_id, 'workflow_link_id': link} + parent['events'][-1] = 'WorkflowContinuedAsNew' + terminal = exports[parent['run_id']]['history_events'][-1] + terminal['type'] = 'WorkflowContinuedAsNew' + terminal['payload'] = {'continued_to_run_id': run_id, 'workflow_link_id': link} + exports[run_id] = { + 'schema': 'durable-workflow.v2.history-export', 'schema_version': 2, 'history_complete': True, + 'workflow': {'run_id': run_id, 'instance_id': workflow_id, + 'source_runtime': 'standalone_server', 'source_process_class': 'server_http_node'}, + 'history_events': events, + } + rows.append({'run_id': run_id, 'workflow_id': workflow_id, 'events': kinds}) + report['run_ids'].append(run_id) + report['cases'][case] = {'command_contract': {'source': 'durable_history'}, + 'runs': rows, 'evidence': {'checked_by_installed_qualifier': True}} + report['c1_run_ids'] = report['run_ids'][:1] + return report, exports + + def test_complete_inventory_binds_all_nine_runs_in_a_read_only_transaction(self): + report, exports = self.fixture() + sql = (RUNNERS / 'condition-wait-anchored-check.sql').read_bytes() + result = gate.sql_input(report, exports, sql) + self.assertIn("SET @run_ids = '" + ','.join(report['run_ids']) + "';", result) + self.assertIn('START TRANSACTION READ ONLY;\n', result) + self.assertIn("SET @expected_namespace = 'php-cwn-0123456789ab';", result) + self.assertIn("SELECT 'R0_run_inventory', COUNT(*)", result) + self.assertIn('HAVING COUNT(*) <> 9;', result) + self.assertTrue(result.endswith('\nROLLBACK;\n')) + + def test_published_cli_progress_is_retained_separately_from_the_final_report(self): + report, _ = self.fixture() + raw = b'ok: runtime advertises condition_wait_idempotent_reopen\nok: history fits one complete page\n' + raw += json.dumps(report, indent=2).encode() + b'\n' + with self.assertRaises(json.JSONDecodeError): + json.loads(raw) + self.assertEqual(gate.extract_report(raw), report) + + def test_cli_noise_crashes_missing_progress_and_partial_reports_refuse(self): + report, _ = self.fixture() + body = json.dumps(report).encode() + for raw in (body, b'ok: started\n', b'unexpected log\n' + body, + b'ok: started\nunexpected log\n' + body, + b'ok: started\n' + body + b'\ntrailing output', + b'ok: started\n{}'): + with self.subTest(raw=raw[:60]): + with self.assertRaises(ValueError): + gate.extract_report(raw) + + def test_missing_duplicate_and_untrusted_run_inventories_refuse(self): + for mutation in ('missing_case', 'extra_case', 'missing_evidence', 'contract', 'duplicate', 'sql_id', 'workflow_id'): + with self.subTest(mutation=mutation): + report, _ = self.fixture() + row = report['cases']['C2']['runs'][0] + if mutation == 'missing_case': del report['cases']['C2'] + if mutation == 'extra_case': report['cases']['unexpected'] = report['cases']['C1'] + if mutation == 'missing_evidence': report['cases']['C2']['evidence'] = {} + if mutation == 'contract': report['cases']['C2']['command_contract']['source'] = 'inferred' + if mutation == 'duplicate': report['run_ids'][1] = report['run_ids'][0] + if mutation == 'sql_id': row['run_id'] = "x'; DELETE FROM workflow_runs; --" + if mutation == 'workflow_id': row['workflow_id'] = '../other-run' + with self.assertRaises(ValueError): gate.report_runs(report) + + def test_incomplete_foreign_imported_and_unbound_exports_refuse(self): + for mutation in ('missing', 'truncated', 'foreign_run', 'foreign_workflow', 'imported', 'gap', + 'duplicate_event', 'different_history', 'wrong_start'): + with self.subTest(mutation=mutation): + report, exports = self.fixture() + run_id = report['run_ids'][0] + export = exports[run_id] + if mutation == 'missing': del exports[run_id] + if mutation == 'truncated': export['history_complete'] = False + if mutation == 'foreign_run': export['workflow']['run_id'] = report['run_ids'][1] + if mutation == 'foreign_workflow': export['workflow']['instance_id'] = 'other-instance' + if mutation == 'imported': export['migration_projection'] = {} + if mutation == 'gap': export['history_events'][1]['sequence'] = 3 + if mutation == 'duplicate_event': export['history_events'][1]['id'] = export['history_events'][0]['id'] + if mutation == 'different_history': export['history_events'][-1]['type'] = 'WorkflowFailed' + if mutation == 'wrong_start': export['history_events'][1]['workflow_command_id'] = 'other-command' + with self.assertRaises(ValueError): gate.validate_exports(report, exports) + + def test_c2_must_actually_overlap_the_activity_before_opening_the_wait(self): + for mutation in ('no_noise', 'before_activity', 'after_activity'): + with self.subTest(mutation=mutation): + report, exports = self.fixture() + run_id = report['cases']['C2']['runs'][0]['run_id'] + events = exports[run_id]['history_events'] + if mutation == 'no_noise': events[4]['payload']['signal_name'] = 'other' + if mutation == 'before_activity': events[2]['type'], events[4]['type'] = events[4]['type'], events[2]['type']; events[2]['payload'] = {'signal_name': 'noise'} + if mutation == 'after_activity': events[4]['type'], events[5]['type'] = events[5]['type'], events[4]['type']; events[5]['payload'] = {'signal_name': 'noise'} + report['cases']['C2']['runs'][0]['events'] = [event['type'] for event in events] + with self.assertRaisesRegex(ValueError, 'signal_not_during_activity'): + gate.validate_exports(report, exports) + + def test_sql_drift_refuses_before_emitting_a_statement(self): + report, exports = self.fixture() + with self.assertRaisesRegex(ValueError, 'sql_identity'): + gate.sql_input(report, exports, (RUNNERS / 'condition-wait-anchored-check.sql').read_bytes() + b'\nSELECT 1;') + + def test_continue_as_new_requires_the_exact_parent_and_both_halves_of_its_link(self): + for mutation in ('parent', 'child', 'link', 'missing_predecessor'): + with self.subTest(mutation=mutation): + report, exports = self.fixture() + parent, child = report['cases']['C5']['runs'] + started = exports[child['run_id']]['history_events'][0]['payload'] + if mutation == 'parent': started['continued_from_run_id'] = report['run_ids'][0] + if mutation == 'child': started['workflow_run_id'] = report['run_ids'][0] + if mutation == 'link': started['workflow_link_id'] = 'other-link' + if mutation == 'missing_predecessor': + export = exports[parent['run_id']] + export['history_events'][-1]['type'] = 'WorkflowCompleted' + parent['events'][-1] = 'WorkflowCompleted' + with self.assertRaisesRegex(ValueError, 'continuation_authority'): + gate.validate_exports(report, exports) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/Unit/Support/php_replay_published_test.py b/tests/Unit/Support/php_replay_published_test.py index 43b60c31a..bcf0bc4c0 100644 --- a/tests/Unit/Support/php_replay_published_test.py +++ b/tests/Unit/Support/php_replay_published_test.py @@ -93,6 +93,7 @@ def test_invalid_artifact_pins_refuse_before_any_docker_command(self): ('DW_SERVER_IMAGE', 'durableworkflow/server:3.3.0'), ('DW_REPLAY_PHP_RUNNER_IMAGE', 'ghcr.io/foreign/tool@sha256:' + 'b' * 64), ('DW_WORKFLOW_LIFECYCLE_MYSQL_IMAGE', 'mysql:latest'), + ('DW_PHP_SDK_CONFORMANCE_CONDITION_WAITS', 'true'), ('DW_JOB_RESOURCE_PREFIX', 'bad/name')]: with self.subTest(key=key): env = dict(base); env[key] = value