diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index 8a8143e63..f2c8a8373 100644 --- a/.github/workflows/server-parity.yml +++ b/.github/workflows/server-parity.yml @@ -323,14 +323,14 @@ jobs: timeout-minutes: 60 strategy: fail-fast: false - # Each database runs in four parts on separate runners: the native + # Each database runs in five parts on separate runners: the native # database suites and TLS checks, the shared corpus with its profiles and - # restart cases, the published-SDK direction profiles, and the separate - # cohorts. Every part bootstraps the same database and native build; no - # part depends on another. + # restart cases, the published-SDK direction profiles, the corrected + # marker cohort, and the other cohorts. Every part bootstraps the same + # database and native build; no part depends on another. matrix: label: [mysql-8.0, existing-php-mysql-matrix, mariadb-10.11] - part: [native-tests, corpus, cohorts, sdk-directions] + part: [native-tests, corpus, cohorts, sdk-directions, corrected-markers] include: - label: mysql-8.0 image: mirror.gcr.io/library/mysql@sha256:62fb722c78b24245ddff1796a0fcee4a49cc5b87e0aaaf20c92d1da9e0a2497b @@ -554,8 +554,17 @@ jobs: sleep 1 done test "$ready" = 1 - docker run --rm --user=1000:1000 --network "$RESOURCE_SCOPE" --entrypoint curl "$image" \ - -fsS http://parity-native-b:8080/api/ready > parity-evidence/native-b-readiness.json + # native-b starts with native-a but may still be binding its port. + ready=0 + for attempt in $(seq 1 30); do + if docker run --rm --user=1000:1000 --network "$RESOURCE_SCOPE" --entrypoint curl "$image" \ + -fsS http://parity-native-b:8080/api/ready > parity-evidence/native-b-readiness.json; then + ready=1 + break + fi + sleep 1 + done + test "$ready" = 1 export DW_PARITY_RUNNER_REVISION=$(git rev-parse HEAD) mkdir -p parity-consumers curl --fail --location --retry 3 --output parity-consumers/dw.phar https://github.com/durable-workflow/cli/releases/download/2.2.0/dw.phar @@ -768,12 +777,38 @@ jobs: DW_PARITY_UPGRADE_EXPECTED_LAYOUT: php-2.9.7 DW_PARITY_UPGRADE_PHP_IMAGE: ghcr.io/durable-workflow/server@sha256:30c34c5f3256a812fbccbe505a068309b58b0c14d01ffff0465d88604cb36d50 run: bash scripts/conformance/server-parity/upgrade-source-cleanup.sh mysql + - name: Start the native node the corrected marker cohort uses + id: prepare_corrected + if: ${{ matrix.part == 'corrected-markers' }} + run: | + set -euo pipefail + image=$(jq -r '.server.images["linux/amd64"] | sub("^durableworkflow/server@"; "ghcr.io/durable-workflow/server@")' tests/Fixtures/ServerParityBaselines/php-2.11.1.json) + docker run --rm --user="$(id -u):$(id -g)" --entrypoint composer \ + -e COMPOSER_HOME=/tmp/composer -v "$PWD:/source" -w /source "$image" \ + install --working-dir=scripts/conformance/server-parity --no-interaction --no-progress + docker run -d --name "$RESOURCE_SCOPE-native-b" --network "$RESOURCE_SCOPE" --network-alias=parity-native-b \ + --user="$(id -u):$(id -g)" --cpus=1 --memory=256m --memory-swap=256m \ + -e DW_RUST_EXPERIMENTAL=1 -e DW_AUTH_TOKEN=parity-test-token -e DW_BIND_ADDRESS=0.0.0.0:8080 \ + -e DB_CONNECTION=mysql -e DB_HOST=parity-mysql -e DB_DATABASE=native_ref \ + -e DB_USERNAME=root -e DB_PASSWORD=parity-disposable-password \ + -v "$PWD/rust-target:/target:ro" --tmpfs /state --entrypoint /target/debug/durable-workflow-server \ + mirror.gcr.io/library/rust@sha256:ba81bc3eaa4422af576c0262515d96b0111a628a6ccc2c86557cf55c9a4bbee0 + ready=0 + for attempt in $(seq 1 30); do + if docker run --rm --user=1000:1000 --network "$RESOURCE_SCOPE" --entrypoint curl "$image" \ + -fsS http://parity-native-b:8080/api/ready > parity-evidence/corrected-native-readiness.json; then + ready=1 + break + fi + sleep 1 + done + test "$ready" = 1 - name: Compare corrected markers, ranges, clocks and activity expiry - if: ${{ !cancelled() && matrix.part == 'cohorts' && steps.prepare_cohorts.outcome == 'success' }} - # The cohorts' own native node; the corpus part kills its native-a. + if: ${{ !cancelled() && matrix.part == 'corrected-markers' && steps.prepare_corrected.outcome == 'success' }} + # The part's own native node; the corpus part kills its native-a. run: bash scripts/conformance/server-parity/corrected-marker-shared-profile.sh mysql http://parity-native-b:8080 - name: Retire corrected cohort resources - if: ${{ always() && matrix.part == 'cohorts' }} + if: ${{ always() && matrix.part == 'corrected-markers' }} run: bash scripts/conformance/server-parity/corrected-marker-cleanup.sh mysql - name: Compare installed local external payload consumers if: ${{ !cancelled() && matrix.part == 'cohorts' }} @@ -899,14 +934,14 @@ jobs: timeout-minutes: 60 strategy: fail-fast: false - # Each database runs in four parts on separate runners: the native + # Each database runs in five parts on separate runners: the native # database suites and TLS checks, the shared corpus with its profiles and - # restart cases, the published-SDK direction profiles, and the separate - # cohorts. Every part bootstraps the same database and native build; no - # part depends on another. + # restart cases, the published-SDK direction profiles, the corrected + # marker cohort, and the other cohorts. Every part bootstraps the same + # database and native build; no part depends on another. matrix: label: [existing-php-matrix, postgres-17] - part: [native-tests, corpus, cohorts, sdk-directions] + part: [native-tests, corpus, cohorts, sdk-directions, corrected-markers] include: - label: existing-php-matrix image: mirror.gcr.io/library/postgres@sha256:f1c3376c26f2609ab9f29f71f824103fe2fcd8ee0346485cb6122a4f93df6f94 @@ -1097,8 +1132,17 @@ jobs: sleep 1 done test "$ready" = 1 - docker run --rm --user=1000:1000 --network "$RESOURCE_SCOPE" --entrypoint curl "$image" \ - -fsS http://parity-native-b:8080/api/ready > parity-evidence/native-b-readiness.json + # native-b starts with native-a but may still be binding its port. + ready=0 + for attempt in $(seq 1 30); do + if docker run --rm --user=1000:1000 --network "$RESOURCE_SCOPE" --entrypoint curl "$image" \ + -fsS http://parity-native-b:8080/api/ready > parity-evidence/native-b-readiness.json; then + ready=1 + break + fi + sleep 1 + done + test "$ready" = 1 export DW_PARITY_RUNNER_REVISION=$(git rev-parse HEAD) mkdir -p parity-consumers curl --fail --location --retry 3 --output parity-consumers/dw.phar https://github.com/durable-workflow/cli/releases/download/2.2.0/dw.phar @@ -1313,14 +1357,40 @@ jobs: DW_PARITY_UPGRADE_EXPECTED_LAYOUT: php-2.9.7 DW_PARITY_UPGRADE_PHP_IMAGE: ghcr.io/durable-workflow/server@sha256:30c34c5f3256a812fbccbe505a068309b58b0c14d01ffff0465d88604cb36d50 run: bash scripts/conformance/server-parity/upgrade-source-cleanup.sh pgsql + - name: Start the native node the corrected marker cohort uses + id: prepare_corrected + if: ${{ matrix.part == 'corrected-markers' }} + run: | + set -euo pipefail + image=$(jq -r '.server.images["linux/amd64"] | sub("^durableworkflow/server@"; "ghcr.io/durable-workflow/server@")' tests/Fixtures/ServerParityBaselines/php-2.11.1.json) + docker run --rm --user="$(id -u):$(id -g)" --entrypoint composer \ + -e COMPOSER_HOME=/tmp/composer -v "$PWD:/source" -w /source "$image" \ + install --working-dir=scripts/conformance/server-parity --no-interaction --no-progress + docker run -d --name "$RESOURCE_SCOPE-native-b" --network "$RESOURCE_SCOPE" --network-alias=parity-native-b \ + --user="$(id -u):$(id -g)" --cpus=1 --memory=256m --memory-swap=256m \ + -e DW_RUST_EXPERIMENTAL=1 -e DW_AUTH_TOKEN=parity-test-token -e DW_BIND_ADDRESS=0.0.0.0:8080 \ + -e DB_CONNECTION=pgsql -e DB_HOST=parity-pg -e DB_DATABASE=native_ref \ + -e DB_USERNAME=parity -e DB_PASSWORD=parity-disposable-password \ + -v "$PWD/rust-target:/target:ro" --tmpfs /state --entrypoint /target/debug/durable-workflow-server \ + mirror.gcr.io/library/rust@sha256:ba81bc3eaa4422af576c0262515d96b0111a628a6ccc2c86557cf55c9a4bbee0 + ready=0 + for attempt in $(seq 1 30); do + if docker run --rm --user=1000:1000 --network "$RESOURCE_SCOPE" --entrypoint curl "$image" \ + -fsS http://parity-native-b:8080/api/ready > parity-evidence/corrected-native-readiness.json; then + ready=1 + break + fi + sleep 1 + done + test "$ready" = 1 - name: Compare corrected markers, ranges, clocks and activity expiry - if: ${{ !cancelled() && matrix.part == 'cohorts' && steps.prepare_cohorts.outcome == 'success' }} + if: ${{ !cancelled() && matrix.part == 'corrected-markers' && steps.prepare_corrected.outcome == 'success' }} env: POSTGRES_IMAGE: ${{ matrix.image }} - # The cohorts' own native node; the corpus part kills its native-a. + # The part's own native node; the corpus part kills its native-a. run: bash scripts/conformance/server-parity/corrected-marker-shared-profile.sh pgsql http://parity-native-b:8080 - name: Retire corrected cohort resources - if: ${{ always() && matrix.part == 'cohorts' }} + if: ${{ always() && matrix.part == 'corrected-markers' }} env: POSTGRES_IMAGE: ${{ matrix.image }} run: bash scripts/conformance/server-parity/corrected-marker-cleanup.sh pgsql @@ -1486,12 +1556,13 @@ jobs: timeout-minutes: 60 strategy: fail-fast: false - # Four parts on separate runners: the native suites and codec, the + # Five parts on separate runners: the native suites and codec, the # shared corpus with its profiles and restart cases, the published-SDK - # direction profiles, and the separate cohorts. Every part builds the - # same native server; no part depends on another. + # direction profiles, the corrected marker cohort, and the other + # cohorts. Every part builds the same native server; no part depends on + # another. matrix: - part: [native-tests, corpus, cohorts, sdk-directions] + part: [native-tests, corpus, cohorts, sdk-directions, corrected-markers] env: DW_PARITY_PART: ${{ matrix.part }} steps: @@ -1598,7 +1669,7 @@ jobs: -e COMPOSER_HOME=/tmp/composer -v "$PWD:/source" -w /source "$image" \ install --working-dir=scripts/conformance/server-parity --no-interaction --no-progress docker run --rm --user=1000:1000 --entrypoint node -v "$PWD:/source:ro" -w /source \ - "$image" --test tests/Unit/ServerParityRunnerTest.mjs tests/Unit/ServerParitySdkDirectionLanesTest.mjs tests/Unit/ServerParityCooperativeContractTest.mjs tests/Unit/ServerParityChildCancellationContractTest.mjs tests/Unit/ServerParityScheduleContractTest.mjs tests/Unit/ServerParityVisibilityContractTest.mjs tests/Unit/ServerParityAdmissionContractTest.mjs tests/Unit/ServerParityRoleTokenContractTest.mjs tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs tests/Unit/ServerParitySearchAttributeContractTest.mjs tests/Unit/ServerParityWorkerListingContractTest.mjs tests/Unit/ServerParityWorkerReplacementContractTest.mjs tests/Unit/ServerParityPatchDeploymentContractTest.mjs tests/Unit/ServerParityWaitingHistoryContractTest.mjs tests/Unit/ServerParityTaskRetryContractTest.mjs tests/Unit/ServerParityPublishedPatchContractTest.mjs tests/Unit/ServerParityPublishedPatchTransportTest.mjs tests/Unit/ServerParityRecordFileTest.mjs tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs tests/Unit/ServerParityStorageTestContractTest.mjs tests/Unit/ServerParitySchedulePoliciesContractTest.mjs tests/Unit/ServerParitySignalsBeforeWorkerContractTest.mjs tests/Unit/ServerParityScheduleCatalogContractTest.mjs tests/Unit/ServerParityOperatorMetricsContractTest.mjs tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs tests/Unit/ServerParityActivityCancellationAckContractTest.mjs tests/Unit/ServerParityServiceCatalogContractTest.mjs tests/Unit/ServerParitySystemMetricsContractTest.mjs tests/Unit/ServerParityOperatorDashboardTypesContractTest.mjs tests/Unit/ServerParityPrometheusMetricsContractTest.mjs tests/Unit/ServerParityPayloadCleanupContractTest.mjs tests/Unit/ServerParityRepairPassContractTest.mjs tests/Unit/ServerParityClusterInfoContractTest.mjs tests/Unit/ServerParityNamespaceDeleteContractTest.mjs tests/Unit/ServerParityMessageStreamsContractTest.mjs tests/Unit/ServerParityWorkflowPollRequestContractTest.mjs tests/Unit/ServerParityUpdateValidationContractTest.mjs tests/Unit/ServerParityFixtureSelectionTest.mjs > parity-evidence/comparator-tests.txt + "$image" --test tests/Unit/ServerParityRunnerTest.mjs tests/Unit/ServerParitySdkDirectionLanesTest.mjs tests/Unit/ServerParityCooperativeContractTest.mjs tests/Unit/ServerParityChildCancellationContractTest.mjs tests/Unit/ServerParityScheduleContractTest.mjs tests/Unit/ServerParityVisibilityContractTest.mjs tests/Unit/ServerParityAdmissionContractTest.mjs tests/Unit/ServerParityRoleTokenContractTest.mjs tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs tests/Unit/ServerParitySearchAttributeContractTest.mjs tests/Unit/ServerParityWorkerListingContractTest.mjs tests/Unit/ServerParityWorkerReplacementContractTest.mjs tests/Unit/ServerParityPatchDeploymentContractTest.mjs tests/Unit/ServerParityWaitingHistoryContractTest.mjs tests/Unit/ServerParityTaskRetryContractTest.mjs tests/Unit/ServerParityPublishedPatchContractTest.mjs tests/Unit/ServerParityPublishedPatchTransportTest.mjs tests/Unit/ServerParityRecordFileTest.mjs tests/Unit/ServerParityFailureLocationTest.mjs tests/Unit/ServerParityFailureExcerptTest.mjs tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs tests/Unit/ServerParityStorageTestContractTest.mjs tests/Unit/ServerParitySchedulePoliciesContractTest.mjs tests/Unit/ServerParitySignalsBeforeWorkerContractTest.mjs tests/Unit/ServerParityScheduleCatalogContractTest.mjs tests/Unit/ServerParityOperatorMetricsContractTest.mjs tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs tests/Unit/ServerParityActivityCancellationAckContractTest.mjs tests/Unit/ServerParityServiceCatalogContractTest.mjs tests/Unit/ServerParitySystemMetricsContractTest.mjs tests/Unit/ServerParityOperatorDashboardTypesContractTest.mjs tests/Unit/ServerParityPrometheusMetricsContractTest.mjs tests/Unit/ServerParityPayloadCleanupContractTest.mjs tests/Unit/ServerParityRepairPassContractTest.mjs tests/Unit/ServerParityClusterInfoContractTest.mjs tests/Unit/ServerParityNamespaceDeleteContractTest.mjs tests/Unit/ServerParityMessageStreamsContractTest.mjs tests/Unit/ServerParityWorkflowPollRequestContractTest.mjs tests/Unit/ServerParityUpdateValidationContractTest.mjs tests/Unit/ServerParityFixtureSelectionTest.mjs > parity-evidence/comparator-tests.txt docker run --rm --user=1000:1000 --cpus=1 --memory=256m --memory-swap=256m \ --network=none --entrypoint node -v "$PWD:/source:ro" -w /source \ "$image" --test tests/Unit/ServerParityCodecDeadlineLoadTest.mjs > parity-evidence/codec-load-tests.txt @@ -1889,13 +1960,46 @@ jobs: DW_PARITY_UPGRADE_EXPECTED_LAYOUT: php-2.9.7 DW_PARITY_UPGRADE_PHP_IMAGE: ghcr.io/durable-workflow/server@sha256:30c34c5f3256a812fbccbe505a068309b58b0c14d01ffff0465d88604cb36d50 run: bash scripts/conformance/server-parity/upgrade-source-cleanup.sh sqlite + - name: Start the native node the corrected marker cohort uses + id: prepare_corrected + if: ${{ matrix.part == 'corrected-markers' }} + env: + RESOURCE_SCOPE: dw-parity-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }} + run: | + set -euo pipefail + image=$(jq -r '.server.images["linux/amd64"] | sub("^durableworkflow/server@"; "ghcr.io/durable-workflow/server@")' tests/Fixtures/ServerParityBaselines/php-2.11.1.json) + docker run --rm --user="$(id -u):$(id -g)" --entrypoint composer \ + -e COMPOSER_HOME=/tmp/composer -v "$PWD:/source" -w /source "$image" \ + install --working-dir=scripts/conformance/server-parity --no-interaction --no-progress + mkdir -p parity-state parity-evidence + # The cohorts' published PHP containers run as 1000:1000; grant that + # user alone, not everyone. + setfacl -m u:1000:rwx,d:u:1000:rwx parity-state parity-evidence + docker network create "$RESOURCE_SCOPE" + docker run -d --name "$RESOURCE_SCOPE-recovered" --network "$RESOURCE_SCOPE" --network-alias=parity-recovered \ + --user="$(id -u):$(id -g)" --cpus=1 --memory=256m --memory-swap=256m \ + -e DW_RUST_EXPERIMENTAL=1 -e DW_AUTH_TOKEN=parity-test-token \ + -e DB_DATABASE=/state/rust.sqlite -e DW_BIND_ADDRESS=0.0.0.0:8080 \ + -v "$PWD/rust-target:/target:ro" -v "$PWD/parity-state:/state" \ + --entrypoint /target/debug/durable-workflow-server \ + mirror.gcr.io/library/rust@sha256:ba81bc3eaa4422af576c0262515d96b0111a628a6ccc2c86557cf55c9a4bbee0 + ready=0 + for attempt in $(seq 1 30); do + if docker run --rm --user=1000:1000 --network "$RESOURCE_SCOPE" --entrypoint curl "$image" \ + -fsS http://parity-recovered:8080/api/ready > parity-evidence/corrected-native-readiness.json; then + ready=1 + break + fi + sleep 1 + done + test "$ready" = 1 - name: Compare corrected markers, ranges, clocks and activity expiry - if: ${{ !cancelled() && matrix.part == 'cohorts' && steps.prepare_cohorts.outcome == 'success' }} + if: ${{ !cancelled() && matrix.part == 'corrected-markers' && steps.prepare_corrected.outcome == 'success' }} env: RESOURCE_SCOPE: dw-parity-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }} run: bash scripts/conformance/server-parity/corrected-marker-shared-profile.sh sqlite http://parity-recovered:8080 - name: Retire corrected cohort resources - if: ${{ always() && matrix.part == 'cohorts' }} + if: ${{ always() && matrix.part == 'corrected-markers' }} env: RESOURCE_SCOPE: dw-parity-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }} run: bash scripts/conformance/server-parity/corrected-marker-cleanup.sh sqlite diff --git a/scripts/conformance/server-parity.mjs b/scripts/conformance/server-parity.mjs index be069ce5a..9fdbd69ca 100644 --- a/scripts/conformance/server-parity.mjs +++ b/scripts/conformance/server-parity.mjs @@ -1,5 +1,5 @@ #!/usr/bin/env node -import {mkdtempSync, readFileSync, rmSync} from 'node:fs'; +import {mkdtempSync, readFileSync, rmSync, writeFileSync} from 'node:fs'; import {createHash} from 'node:crypto'; import {fileURLToPath} from 'node:url'; import {basename, dirname, join, resolve} from 'node:path'; @@ -20,6 +20,8 @@ import {checkCorrectedMarkerArtifacts, checkCorrectedMarkerPackages} from './ser import {checkCorrectedRangeArtifacts, checkCorrectedRangePackages} from './server-parity/corrected-range-contract.mjs'; import {checkActivityDeadlinePackages} from './server-parity/activity-deadline-contract.mjs'; import {checkActivityDeadlineReceiptPackages} from './server-parity/activity-deadline-receipts-contract.mjs'; +import {contractFailureLocation} from './server-parity/failure-location.mjs'; +import {failureExcerpt} from './server-parity/failure-excerpt.mjs'; const root = resolve(dirname(fileURLToPath(import.meta.url)), '../..'); const {values, positionals} = parseArgs({allowPositionals: true, options: { @@ -139,7 +141,8 @@ try { // its own workflow. Report it and keep recording, so one run shows // every difference. A runner-blocked case still stops the record. if (item.outcome === 'product-fail') { - console.error(`${fixture.id}: ${error.message}`); + const location = contractFailureLocation(error); + console.error(`${fixture.id}: ${error.message}${location ? `\n at ${location}` : ''}`); continue; } record.outcome = item.outcome; @@ -155,6 +158,9 @@ try { record.outcome = 'product-fail'; record.finished_at = new Date().toISOString(); writeRecord(values.output, record, originals); + // The artifact summary keeps this outline; it does not keep the record. + writeFileSync(values.output.replace(/\.json$/, '') + '.failure-excerpt.txt', + failureExcerpt(record.cases.filter(item => item.outcome !== 'pass'))); throw new Error(`${failed.length} fixture(s) differ: ${failed.join(', ')}`); } record.outcome = 'pass'; diff --git a/scripts/conformance/server-parity/artifact-summary.mjs b/scripts/conformance/server-parity/artifact-summary.mjs index 91b58da86..a2a24cac3 100644 --- a/scripts/conformance/server-parity/artifact-summary.mjs +++ b/scripts/conformance/server-parity/artifact-summary.mjs @@ -12,7 +12,7 @@ function priority(path, failed) { if (failed && /(?:^|\/)query-authority\/clients\/(?:php|rust)\/[^/]+\/(?:client\.stderr|observer\.log)$/.test(path)) return 0; if (/^(rust|mysql|postgres|comparator|codec-load|artifact-summary)(-execution)?-tests\.txt$/.test(name)) return 0; if (/^(comparison|compare)([-.].*)?\.json$/.test(name) || /-comparison\.json$/.test(name)) return 1; - if (failed && (/^(?:php|embedded|rust)-record\.txt$/.test(name) + if (failed && (/^(?:php|embedded|rust)-record\.txt$/.test(name) || /\.failure-excerpt\.txt$/.test(name) || /^sdk-(?:signals|queries)-profile\.log$/.test(name))) return 1; if (failed && (/\.stderr$/.test(name) || /(?:stderr|error|failure|bootstrap|install|build).*(?:\.log|\.txt)$/.test(name) || /^(?:php|rust|native)(?:[-.].*)?\.log$/.test(name) || /server\.log$/.test(name))) return 2; diff --git a/scripts/conformance/server-parity/failure-excerpt.mjs b/scripts/conformance/server-parity/failure-excerpt.mjs new file mode 100644 index 000000000..61a143c83 --- /dev/null +++ b/scripts/conformance/server-parity/failure-excerpt.mjs @@ -0,0 +1,52 @@ +// A bounded, value-free outline of every wire request a failed case observed, +// in observation order, so a flake can be diagnosed from the retained artifact +// without keeping the record. It names each request's method, path, status, +// transport flags, worker and poll identity. Request and response bodies, +// payloads and credentials are never read into it. + +const SKIP = new Set(['request', 'response', 'request_body', 'response_body', 'response_bytes_base64', + 'transport_jsonl', 'stdout', 'stderr', 'fixture']); + +function line(path, item) { + const fields = [`${path} ${item.method} ${item.path} ${item.status}`]; + for (const key of ['client_cancelled', 'transport_error', 'request_body_complete', 'upstream_request_started', + 'response_retry_after']) { + if (Object.hasOwn(item, key) && item[key] !== null && item[key] !== false) fields.push(`${key}=${item[key]}`); + } + const request = item.request && typeof item.request === 'object' ? item.request : {}; + for (const key of ['worker_id', 'poll_request_id', 'task_queue', 'lease_owner']) { + if (typeof request[key] === 'string') fields.push(`${key}=${request[key].slice(0, 120)}`); + } + return fields.join(' '); +} + +function walk(value, path, out, limit) { + if (out.length >= limit || value === null || typeof value !== 'object') return; + if (Array.isArray(value)) { + value.forEach((item, index) => walk(item, `${path}[${index}]`, out, limit)); + return; + } + if (typeof value.method === 'string' && typeof value.path === 'string' && Object.hasOwn(value, 'status')) { + out.push(line(path, value)); + } + for (const [key, child] of Object.entries(value)) { + if (!SKIP.has(key)) walk(child, `${path}.${key}`, out, limit); + } +} + +export function failureExcerpt(cases, {linesPerCase = 300, maxBytes = 60 * 1024} = {}) { + let text = ''; + for (const item of cases) { + const lines = []; + walk(item.observation, 'observation', lines, linesPerCase + 1); + const truncated = lines.length > linesPerCase; + const block = [`== ${item.fixture_id} (${item.outcome}): ${String(item.error ?? '').split('\n')[0].slice(0, 300)}`, + ...lines.slice(0, linesPerCase), ...(truncated ? [`… more requests omitted after ${linesPerCase}`] : []), ''].join('\n'); + if (Buffer.byteLength(text + block) > maxBytes) { + text += `… further failed cases omitted at ${maxBytes} bytes\n`; + break; + } + text += block; + } + return text; +} diff --git a/scripts/conformance/server-parity/failure-location.mjs b/scripts/conformance/server-parity/failure-location.mjs new file mode 100644 index 000000000..1891d7c98 --- /dev/null +++ b/scripts/conformance/server-parity/failure-location.mjs @@ -0,0 +1,10 @@ +// The contract line that refused a fixture, as "scripts/conformance/...:line". +// A bare "false !== true" is not locatable from a retained log; the stack +// frame is. Frames carry source paths and line numbers, never payload values. +export function contractFailureLocation(error) { + for (const frame of String(error?.stack ?? '').split('\n').slice(1)) { + const match = frame.match(/(scripts\/conformance\/[^\s:()]+\.m?js):(\d+):\d+/); + if (match) return `${match[1]}:${match[2]}`; + } + return null; +} diff --git a/scripts/conformance/server-parity/sdk-query-direction-contract.mjs b/scripts/conformance/server-parity/sdk-query-direction-contract.mjs index 731b6051e..b7aaa35af 100644 --- a/scripts/conformance/server-parity/sdk-query-direction-contract.mjs +++ b/scripts/conformance/server-parity/sdk-query-direction-contract.mjs @@ -35,6 +35,15 @@ const one = (requests, predicate, message) => { assert.equal(selected.length, 1, message); return selected[0]; }; +// A wrong status names the request and its observer transport flags, never +// payload bytes. Status 0 means no upstream status arrived, and the flags say +// whether the client closed first or the upstream connection failed. +function wireStatus(item, expected, label) { + if (item.status === expected) return; + const {method, path, status, client_cancelled, transport_error, request_body_complete, upstream_request_started} = item; + assert.fail(`${label}: expected HTTP ${expected}, observed ${status} for ${method} ${path} ` + + JSON.stringify({client_cancelled, transport_error, request_body_complete, upstream_request_started})); +} function unchanged(before, after, message) { assert.deepEqual(after, before, message); } @@ -144,7 +153,7 @@ export function checkSdkQueryDirection(fixture, observation, workflowId) { assert(!phase.requests.some(item => item.method === 'POST' && item.path === '/api/workflows')); const wire = one(phase.requests, item => item.method === 'POST' && item.path === '/api/workflows/' + workflowId + '/signal/payload', 'one actual signal stimulus'); - assert.equal(wire.status, 202); + wireStatus(wire, 202, phase.phase + ' signal'); assert.equal(wire.response.run_id, observation.run_id); assert.equal(wire.response.workflow_id, workflowId); assert.equal(wire.response.outcome, 'signal_received'); @@ -165,7 +174,7 @@ export function checkSdkQueryDirection(fixture, observation, workflowId) { unchangedDurableSnapshot(phase.snapshot_before, direction.phases[index - 1].snapshot, 'same acknowledged author snapshot'); const wire = one(phase.requests, item => item.method === 'POST' && item.path === '/api/workflows/' + workflowId + '/query/state', 'one normal instance-scoped query'); - assert.equal(wire.status, 200); + wireStatus(wire, 200, phase.phase + ' query'); assert.equal(wire.response.workflow_id, workflowId); assert.equal(wire.response.run_id, observation.run_id); assert.equal(wire.response.query_name, 'state'); @@ -188,7 +197,7 @@ export function checkSdkQueryDirection(fixture, observation, workflowId) { const claim = one(author.requests, item => item.path === '/api/worker/query-tasks/poll' && item.response?.task, 'one actual query worker claim'); const task = claim.response.task; - assert.equal(claim.method, 'POST'); assert.equal(claim.status, 200); + assert.equal(claim.method, 'POST'); wireStatus(claim, 200, author.phase + ' query claim'); assert.equal(claim.request.task_queue, spec.task_queue); assert.equal(task.workflow_id, workflowId); assert.equal(task.run_id, observation.run_id); assert.equal(task.workflow_type, fixture.workflow_type); assert.equal(task.query_name, 'state'); @@ -211,7 +220,7 @@ export function checkSdkQueryDirection(fixture, observation, workflowId) { if (wire.request.input) assert.equal(frame(wire.request.input), frame(task.query_arguments)); const complete = one(author.requests, item => item.method === 'POST' && item.path === '/api/worker/query-tasks/' + task.query_task_id + '/complete', 'actual original query completion'); - assert.equal(complete.status, 200); + wireStatus(complete, 200, author.phase + ' query completion'); assert.equal(complete.request.lease_owner, task.lease_owner); assert.equal(complete.request.query_task_attempt, task.query_task_attempt); assert.equal(complete.response.query_task_id, task.query_task_id); @@ -232,7 +241,7 @@ export function checkSdkQueryDirection(fixture, observation, workflowId) { if (language === 'php') assert.equal(phase.query_completions, 1); const register = one(phase.requests, item => item.method === 'POST' && item.path === '/api/worker/register', 'one ordinary worker registration'); - assert.equal(register.status, 201); assert.equal(register.response.registered, true); + wireStatus(register, 201, phase.phase + ' registration'); assert.equal(register.response.registered, true); assert.equal(register.request.worker_id, phase.worker_id); assert.equal(register.request.task_queue, spec.task_queue); assert.deepEqual(register.request.supported_workflow_types, [fixture.workflow_type]); @@ -241,7 +250,7 @@ export function checkSdkQueryDirection(fixture, observation, workflowId) { const claim = one(phase.requests, item => item.path === '/api/worker/workflow-tasks/poll' && item.response?.task, 'one original workflow claim per cold author'); const task = claim.response.task; - assert.equal(claim.status, 200); + wireStatus(claim, 200, phase.phase + ' workflow claim'); for (const key of ['workflow_id', 'run_id', 'workflow_type', 'task_queue']) assert.equal(task[key], observation[key]); assert.equal(task.workflow_task_attempt, 1); nonempty(task.task_id); nonempty(task.lease_owner); checkWorkerHistory(task.history_events); @@ -251,7 +260,7 @@ export function checkSdkQueryDirection(fixture, observation, workflowId) { assert.deepEqual(history(task.history_events), history(observation.events.slice(0, [2, 4, 9][turn]))); const complete = one(phase.requests, item => item.method === 'POST' && item.path === '/api/worker/workflow-tasks/' + task.task_id + '/complete', 'original fenced workflow completion'); - assert.equal(complete.status, 200); assert.equal(complete.response.recorded, true); + wireStatus(complete, 200, phase.phase + ' workflow completion'); assert.equal(complete.response.recorded, true); assert.equal(complete.response.task_id, task.task_id); assert.equal(complete.response.run_id, observation.run_id); assert.equal(complete.request.lease_owner, task.lease_owner); assert.equal(complete.request.workflow_task_attempt, task.workflow_task_attempt); @@ -264,7 +273,7 @@ export function checkSdkQueryDirection(fixture, observation, workflowId) { if (turn === 0) { const start = one(phase.requests, item => item.method === 'POST' && item.path === '/api/workflows', 'one real observer start'); - assert.equal(start.status, 201); + wireStatus(start, 201, phase.phase + ' start'); assert.equal(start.request.workflow_id, workflowId); assert.equal(start.response.run_id, observation.run_id); assert(phase.requests.indexOf(register) < phase.requests.indexOf(start)); assert(phase.requests.indexOf(start) < phase.requests.indexOf(claim), 'original first claim after successful start receipt'); diff --git a/scripts/conformance/server-parity/sdk-query-direction-probe.php b/scripts/conformance/server-parity/sdk-query-direction-probe.php index 1beb8c232..dd7ca7881 100644 --- a/scripts/conformance/server-parity/sdk-query-direction-probe.php +++ b/scripts/conformance/server-parity/sdk-query-direction-probe.php @@ -191,6 +191,35 @@ function sdkQueryWorkflowAcknowledgement(array $requests, string $workflowId, st return null; } +/** + * True once the author's own completion of a query task it claimed for this + * run has a successful, uncancelled response in its journal. + */ +function sdkQueryCompletionAcknowledged(array $requests, string $workflowId, string $runId): bool +{ + foreach ($requests as $pollIndex => $poll) { + $task = is_array($poll['response'] ?? null) ? ($poll['response']['task'] ?? null) : null; + if (($poll['method'] ?? null) !== 'POST' || ($poll['path'] ?? null) !== '/api/worker/query-tasks/poll' + || ($poll['status'] ?? null) !== 200 || ! is_array($task) + || ($task['workflow_id'] ?? null) !== $workflowId || ($task['run_id'] ?? null) !== $runId + || ! is_string($task['query_task_id'] ?? null) || $task['query_task_id'] === '') { + continue; + } + $path = '/api/worker/query-tasks/'.$task['query_task_id'].'/complete'; + foreach ($requests as $index => $completion) { + if ($index > $pollIndex && ($completion['method'] ?? null) === 'POST' && ($completion['path'] ?? null) === $path + && ($completion['status'] ?? null) === 200 && ($completion['client_cancelled'] ?? true) === false + && ($completion['transport_error'] ?? null) === null + && is_array($completion['response'] ?? null) + && ($completion['response']['query_task_id'] ?? null) === $task['query_task_id']) { + return true; + } + } + } + + return false; +} + function sdkQueryActorFinish(array &$actor, float $deadline, string $evidence, string $workflowId): array { while (sdkQueryActorPoll($actor)) { @@ -307,6 +336,15 @@ function sdkQueryDirectionHttpObservation(array $fixture, array $options): array $query = sdkQueryActorFinish($actors[$queryPhase], $deadline, $evidence, $options['workflow-id']); $after = sdkDirectionSnapshot($client, $options['workflow-id'], $runId); if (($query['client_finished'] ?? false) === true) { + // Native wakes the caller as soon as the completion commits, so + // the caller can finish before the author's own completion + // response arrives. Stopping the author then cancels that request + // (published Python runs queries on a thread its stop cancels). + // Release the author only after its completion is observed. + while (! sdkQueryCompletionAcknowledged(sdkQueryLiveWireRequests($author), $options['workflow-id'], $runId) + && sdkQueryActorPoll($author) && microtime(true) < $deadline) { + usleep(10000); + } file_put_contents($author['receipt'], $actors[$queryPhase]['stdout']); } else { proc_terminate($author['process']); diff --git a/tests/Unit/ServerParityArtifactSummaryTest.mjs b/tests/Unit/ServerParityArtifactSummaryTest.mjs index 3dbe14f49..7ec275f81 100644 --- a/tests/Unit/ServerParityArtifactSummaryTest.mjs +++ b/tests/Unit/ServerParityArtifactSummaryTest.mjs @@ -134,3 +134,20 @@ test('failed recorder diagnostic tails survive without retaining raw records or summarizeEvidence(source, success, {outcome: 'success'}); assert.equal(existsSync(join(success, 'php-record.txt')), false); }); + +test('a failed record outline survives a full collection of setup logs and stays out of success summaries', t => { + const {source, output, root} = fixture(t); + for (let index = 0; index < 140; index++) { + writeFileSync(join(source, `completed-${index}-install.txt`), 'installed successfully\n'); + } + mkdirSync(join(source, 'legacy-marker/php'), {recursive: true}); + writeFileSync(join(source, 'legacy-marker/php/record.failure-excerpt.txt'), '== legacy-one-marker-control (product-fail)\n'); + writeFileSync(join(source, 'legacy-marker/php/record.json'), 'raw responses'); + summarizeEvidence(source, output, {outcome: 'failure'}); + assert.equal(readFileSync(join(output, 'legacy-marker/php/record.failure-excerpt.txt'), 'utf8'), + '== legacy-one-marker-control (product-fail)\n'); + assert.equal(existsSync(join(output, 'legacy-marker/php/record.json')), false); + const success = join(root, 'success'); + summarizeEvidence(source, success, {outcome: 'success'}); + assert.equal(existsSync(join(success, 'legacy-marker/php/record.failure-excerpt.txt')), false); +}); diff --git a/tests/Unit/ServerParityFailureExcerptTest.mjs b/tests/Unit/ServerParityFailureExcerptTest.mjs new file mode 100644 index 000000000..38417c57a --- /dev/null +++ b/tests/Unit/ServerParityFailureExcerptTest.mjs @@ -0,0 +1,39 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import {failureExcerpt} from '../../scripts/conformance/server-parity/failure-excerpt.mjs'; + +const poll = (status, extra = {}) => ({method: 'POST', path: '/api/worker/workflow-tasks/poll', status, + request: {worker_id: 'w1', poll_request_id: 'p1', task_queue: 'q', secret_input: 'payload-bytes'}, + response: {reason: 'lock pressure', nested: {method: 'GET', path: '/not-a-request', status: 200}}, + request_body: '{"secret_input":"payload-bytes"}', response_body: 'body-bytes', client_cancelled: false, + transport_error: null, ...extra}); + +test('a failed case lists its requests in order with identity and transport flags, never bodies', () => { + const text = failureExcerpt([{fixture_id: 'legacy-one-marker-control', outcome: 'product-fail', + error: 'same original poll recovers\nsecond line', observation: {legacy_marker: {phases: [ + {role: 'original', requests: [poll(503, {response_retry_after: '2'}), poll(200)]}, + {role: 'replacement', requests: [{method: 'DELETE', path: '/api/worker/registrations/w1', status: 200, + request: null, client_cancelled: true}]}]}}}]); + assert.equal(text, [ + '== legacy-one-marker-control (product-fail): same original poll recovers', + 'observation.legacy_marker.phases[0].requests[0] POST /api/worker/workflow-tasks/poll 503 response_retry_after=2 worker_id=w1 poll_request_id=p1 task_queue=q', + 'observation.legacy_marker.phases[0].requests[1] POST /api/worker/workflow-tasks/poll 200 worker_id=w1 poll_request_id=p1 task_queue=q', + 'observation.legacy_marker.phases[1].requests[0] DELETE /api/worker/registrations/w1 200 client_cancelled=true', + ''].join('\n')); + assert(!text.includes('payload-bytes') && !text.includes('body-bytes') && !text.includes('/not-a-request')); +}); + +test('the outline is bounded per case and in total', () => { + const many = Array.from({length: 50}, () => poll(200)); + const one = failureExcerpt([{fixture_id: 'a', outcome: 'product-fail', observation: {requests: many}}], {linesPerCase: 10}); + assert.equal(one.split('\n').filter(l => l.startsWith('observation')).length, 10); + assert.match(one, /more requests omitted after 10/); + const cases = Array.from({length: 20}, (_, i) => ({fixture_id: `case-${i}`, outcome: 'product-fail', observation: {requests: many}})); + const all = failureExcerpt(cases, {maxBytes: 8 * 1024}); + assert(Buffer.byteLength(all) <= 8 * 1024 + 100); + assert.match(all, /further failed cases omitted at 8192 bytes/); +}); + +test('a case without an observation still names its error', () => { + assert.equal(failureExcerpt([{fixture_id: 'x', outcome: 'runner-blocked', error: 'boom'}]), '== x (runner-blocked): boom\n'); +}); diff --git a/tests/Unit/ServerParityFailureLocationTest.mjs b/tests/Unit/ServerParityFailureLocationTest.mjs new file mode 100644 index 000000000..3f263a664 --- /dev/null +++ b/tests/Unit/ServerParityFailureLocationTest.mjs @@ -0,0 +1,28 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import {contractFailureLocation} from '../../scripts/conformance/server-parity/failure-location.mjs'; +import {checkSdkQueryDirection} from '../../scripts/conformance/server-parity/sdk-query-direction-contract.mjs'; + +test('a contract refusal names its source file and line', () => { + let refusal; + try { + checkSdkQueryDirection({sdk_query_direction: {}}, {mode: 'http', sdk_query_direction: {applicable: true}}, 'w'); + } catch (error) { refusal = error; } + assert.ok(refusal, 'the malformed observation is refused'); + assert.match(contractFailureLocation(refusal), /^scripts\/conformance\/server-parity\/[a-z-]+\.mjs:\d+$/); +}); + +test('container and file URL frames resolve to the same repository path', () => { + for (const frame of [' at check (/source/scripts/conformance/server-parity/x-contract.mjs:42:7)', + ' at check (file:///source/scripts/conformance/server-parity/x-contract.mjs:42:7)', + ' at file:///home/runner/work/server/server/scripts/conformance/server-parity/x-contract.mjs:42:7']) { + assert.equal(contractFailureLocation({stack: 'AssertionError: false !== true\n' + frame}), + 'scripts/conformance/server-parity/x-contract.mjs:42'); + } +}); + +test('a failure outside the contracts has no location rather than a guess', () => { + assert.equal(contractFailureLocation({stack: 'Error: x\n at node:internal/process/task_queues:95:5'}), null); + assert.equal(contractFailureLocation('not an error'), null); + assert.equal(contractFailureLocation(undefined), null); +}); diff --git a/tests/Unit/ServerParityPartsTest.mjs b/tests/Unit/ServerParityPartsTest.mjs index 5a8a81b82..8d953121d 100644 --- a/tests/Unit/ServerParityPartsTest.mjs +++ b/tests/Unit/ServerParityPartsTest.mjs @@ -1,4 +1,4 @@ -// Each Shared Server fixtures database job runs in four parts on separate +// Each Shared Server fixtures database job runs in five parts on separate // runners. Every step either prepares the part (and so runs in every part) or // belongs to exactly one part, so no check runs twice or not at all. import test from 'node:test'; @@ -6,7 +6,7 @@ import assert from 'node:assert/strict'; import {readFileSync} from 'node:fs'; const workflow = readFileSync('.github/workflows/server-parity.yml', 'utf8'); -const PARTS = ['native-tests', 'corpus', 'cohorts', 'sdk-directions']; +const PARTS = ['native-tests', 'corpus', 'cohorts', 'sdk-directions', 'corrected-markers']; const EVERY_PART = [/^actions\/checkout@/, /^Fetch /, /^Create TLS fixture certificates$/, /^Bootstrap independent /, /^Collect /, /^Summarize /, /^Retain /]; @@ -39,9 +39,9 @@ function partsOf(step) { for (const [name, key] of [['mysql-execution', 'Compare three runtimes'], ['postgres-storage', 'Compare three runtimes'], ['execution-and-codec', 'Execute and compare Rust']]) { - test(`${name} runs every check in exactly one of four parts`, () => { + test(`${name} runs every check in exactly one of five parts`, () => { const block = job(name); - assert.match(block, /^ {8}part: \[native-tests, corpus, cohorts, sdk-directions\]$/m); + assert.match(block, /^ {8}part: \[native-tests, corpus, cohorts, sdk-directions, corrected-markers\]$/m); assert.match(block, /^ {6}DW_PARITY_PART: \$\{\{ matrix\.part \}\}$/m); assert.match(block, /^ {4}timeout-minutes: 60$/m); const all = steps(block); @@ -64,19 +64,24 @@ for (const [name, key] of [['mysql-execution', 'Compare three runtimes'], ['post for (const cohort of ['Start the shared native node the cohorts use', 'Compare separate configured role-token cohort', 'Back up, take over and resume a separate published PHP upgrade source', 'Back up, take over and resume a published PHP 2.9.7 upgrade source', - 'Compare corrected markers, ranges, clocks and activity expiry', 'Compare installed local external payload consumers', + 'Compare installed local external payload consumers', 'Compare published query deadlines and installed Worker authority', 'Compare pinned PHP, embedded and native worker-fencing source cohort', 'Compare SDK temporary database pressure with declared fault limits']) { assert.ok(cohorts.includes(cohort), `${name}: ${cohort} is in the cohorts part`); } + // The corrected marker cohort runs alone in its part, against that part's + // own native node, so the cohorts part no longer waits for it. + assert.deepEqual(inPart('corrected-markers'), ['Start the native node the corrected marker cohort uses', + 'Compare corrected markers, ranges, clocks and activity expiry', 'Retire corrected cohort resources'], name); // Cleanups still run after a failure within their part. for (const step of all.filter(step => step.name.startsWith('Retire '))) { - assert.match(step.condition, /^\$\{\{ always\(\) && matrix\.part == 'cohorts' \}\}$/, step.name); + const part = step.name === 'Retire corrected cohort resources' ? 'corrected-markers' : 'cohorts'; + assert.equal(step.condition, `\${{ always() && matrix.part == '${part}' }}`, step.name); } - // The corrected cohort runs against the cohorts' own native node. const corrected = all.find(step => step.name.startsWith('Compare corrected markers')); - assert.match(corrected.condition, /steps\.prepare_cohorts\.outcome == 'success'/); + assert.match(corrected.condition, /steps\.prepare_corrected\.outcome == 'success'/); + assert.equal(all.find(step => step.name === 'Start the native node the corrected marker cohort uses').id, 'prepare_corrected'); assert.doesNotMatch(block, /steps\.prepare_runtime_comparison\.outcome/); assert.equal(all.find(step => step.name === 'Start the shared native node the cohorts use').id, 'prepare_cohorts'); // Each part retains its own summary. diff --git a/tests/Unit/ServerParitySdkQueryDirectionContractTest.mjs b/tests/Unit/ServerParitySdkQueryDirectionContractTest.mjs index c314b62ea..4a21b050f 100644 --- a/tests/Unit/ServerParitySdkQueryDirectionContractTest.mjs +++ b/tests/Unit/ServerParitySdkQueryDirectionContractTest.mjs @@ -223,3 +223,21 @@ test('accept explicit null reason on successful native query completion', () => assert.deepEqual(checkSdkQueryDirection(fixture, observation, 'query-model'), {three_fresh_readonly_queries: true, exact_typed_completion: true}); }); +test('a status-0 request names its path and transport flags, never payload bytes', () => { + const observation = model(fixtureFor('python', 'rust')); + Object.assign(observation.sdk_query_direction.phases[1].requests[0], {status: 0, response: null, + client_cancelled: true, transport_error: null, request_body_complete: true, upstream_request_started: true}); + assert.throws(() => checkSdkQueryDirection(fixtureFor('python', 'rust'), observation, 'query-model'), error => { + assert.equal(error.message, 'query-0 query: expected HTTP 200, observed 0 for POST /api/workflows/query-model/query/state ' + + '{"client_cancelled":true,"transport_error":null,"request_body_complete":true,"upstream_request_started":true}'); + assert(!error.message.includes('empty-arguments'), 'request payload stays out of the message'); + return true; + }); +}); +test('a failed query completion names the completion, not the caller', () => { + const observation = model(fixture); + const complete = observation.sdk_query_direction.phases[0].requests.find(r => /query-tasks\/.*\/complete/.test(r.path)); + Object.assign(complete, {status: 0, response: null, client_cancelled: false, transport_error: 'ECONNRESET'}); + assert.throws(() => checkSdkQueryDirection(fixture, observation, 'query-model'), + /original query completion: expected HTTP 200, observed 0 for POST \/api\/worker\/query-tasks\/.*"transport_error":"ECONNRESET"/); +}); diff --git a/tests/Unit/ServerParitySdkQueryTransportTest.php b/tests/Unit/ServerParitySdkQueryTransportTest.php index 021e19db5..56c176f3e 100644 --- a/tests/Unit/ServerParitySdkQueryTransportTest.php +++ b/tests/Unit/ServerParitySdkQueryTransportTest.php @@ -252,4 +252,48 @@ public function test_a_live_journal_cannot_lose_already_committed_bytes(): void unlink($journal); } } + + private function queryCompletionRequests(): array + { + return [ + ['method' => 'POST', 'path' => '/api/worker/query-tasks/poll', 'status' => 200, + 'client_cancelled' => false, 'transport_error' => null, + 'response' => ['task' => ['query_task_id' => 'query-task', 'workflow_id' => 'original', 'run_id' => 'run']]], + ['method' => 'POST', 'path' => '/api/worker/query-tasks/query-task/complete', 'status' => 200, + 'client_cancelled' => false, 'transport_error' => null, + 'response' => ['query_task_id' => 'query-task', 'outcome' => 'completed']], + ]; + } + + public function test_the_author_is_released_only_after_its_query_completion_response(): void + { + self::assertTrue(\sdkQueryCompletionAcknowledged($this->queryCompletionRequests(), 'original', 'run')); + self::assertFalse(\sdkQueryCompletionAcknowledged([$this->queryCompletionRequests()[0]], 'original', 'run'), + 'a claimed query whose completion has no response yet keeps the author running'); + } + + public function test_cancelled_failed_foreign_or_misordered_query_completions_do_not_release_the_author(): void + { + // The observed flake: the caller returned, the author was stopped, and + // its in-flight completion was recorded as client-cancelled status 0. + foreach ([ + 'status' => 0, 'client_cancelled' => true, 'transport_error' => 'ECONNRESET', + ] as $field => $value) { + $requests = $this->queryCompletionRequests(); + $requests[1][$field] = $value; + self::assertFalse(\sdkQueryCompletionAcknowledged($requests, 'original', 'run'), $field); + } + $requests = $this->queryCompletionRequests(); + $requests[1]['response'] = new stdClass; + self::assertFalse(\sdkQueryCompletionAcknowledged($requests, 'original', 'run'), 'an empty response object'); + $requests = $this->queryCompletionRequests(); + $requests[1]['response']['query_task_id'] = 'foreign-query'; + self::assertFalse(\sdkQueryCompletionAcknowledged($requests, 'original', 'run'), 'a foreign completion receipt'); + $requests = $this->queryCompletionRequests(); + $requests[0]['response'] = new stdClass; + self::assertFalse(\sdkQueryCompletionAcknowledged($requests, 'original', 'run'), 'an empty poll'); + self::assertFalse(\sdkQueryCompletionAcknowledged(array_reverse($this->queryCompletionRequests()), 'original', 'run')); + self::assertFalse(\sdkQueryCompletionAcknowledged($this->queryCompletionRequests(), 'foreign', 'run')); + self::assertFalse(\sdkQueryCompletionAcknowledged($this->queryCompletionRequests(), 'original', 'foreign')); + } }