diff --git a/scripts/ci/run-composer.sh b/scripts/ci/run-composer.sh index ed48f33d7..528dbcb5b 100644 --- a/scripts/ci/run-composer.sh +++ b/scripts/ci/run-composer.sh @@ -36,12 +36,33 @@ shift image=$(sh "$here/brotli-safe-composer.sh") ca=$(mktemp -d "${RUNNER_TEMP:-${TMPDIR:-/tmp}}/dw-composer-ca.XXXXXX") -trap 'rm -rf -- "$ca"' EXIT +errors=$(mktemp "${RUNNER_TEMP:-${TMPDIR:-/tmp}}/dw-composer-stderr.XXXXXX") +trap 'rm -rf -- "$ca" "$errors"' EXIT sh "$here/checked-ca.sh" "$ca" chmod 0755 "$ca" -status=0 -docker run --user "$(id -u):$(id -g)" -e COMPOSER_HOME=/tmp/composer "${options[@]}" \ - -e COMPOSER_CAFILE=/checked-ca/ca.pem -e COMPOSER_CAPATH=/checked-ca/no-ca-directory \ - -v "$ca:/checked-ca:ro" --entrypoint composer "$image" "$@" || status=$? +# A locked install only downloads pinned dists, so it is safe to repeat. Retry +# it, at most three attempts in all, only when Composer reports a transient +# download failure such as GitHub codeload's HTTP 429 rate limit. Any other +# failure, every other Composer command and every kept (named, not --rm) +# container fails on its first attempt. +transient='could not be downloaded \((HTTP/[0-9.]+ (429|5[0-9][0-9])|curl error (6|7|28|35|52|56))' +attempts=1 +if [[ $1 == install ]] && [[ " ${options[*]} " == *" --rm "* ]]; then attempts=3; fi +for attempt in $(seq "$attempts"); do + # Composer's stderr reaches the caller unchanged and is kept for the check. + set +o errexit + { docker run --user "$(id -u):$(id -g)" -e COMPOSER_HOME=/tmp/composer "${options[@]}" \ + -e COMPOSER_CAFILE=/checked-ca/ca.pem -e COMPOSER_CAPATH=/checked-ca/no-ca-directory \ + -v "$ca:/checked-ca:ro" --entrypoint composer "$image" "$@" 2>&1 1>&3 3>&- | tee "$errors" >&2 3>&-; } 3>&1 + statuses=("${PIPESTATUS[@]}") + set -o errexit + status=${statuses[0]} + ((statuses[1] == 0)) || status=${statuses[1]} + if ((status == 0 || attempt == attempts)) || ! grep -Eq "$transient" "$errors"; then + break + fi + echo "run-composer.sh: transient download failure on attempt $attempt of $attempts; retrying in $((attempt * 20))s" >&2 + sleep $((attempt * 20)) +done exit "$status" diff --git a/scripts/conformance/server-parity/patch-deployment-contract.mjs b/scripts/conformance/server-parity/patch-deployment-contract.mjs index a0ca102e2..f13c32c77 100644 --- a/scripts/conformance/server-parity/patch-deployment-contract.mjs +++ b/scripts/conformance/server-parity/patch-deployment-contract.mjs @@ -10,6 +10,44 @@ export function checkPatchPollPressureResponse(response) { if (Object.hasOwn(response, 'message')) assert.equal(typeof response.message, 'string', 'published diagnostic message is text'); } +// A server whose worker long-poll wait slots are full answers a workflow or +// activity poll with PHP's typed, retryable 429 (WorkerPollBackpressure, the +// DW_WORKER_LONG_POLL_MAX_CONCURRENT contract) before holding a wait. It +// leases nothing. The worker then retries that same poll or, with its work +// done, deregisters; either way no task, outcome or authority is lost. +function checkLongPollCapacity(item, index, phase, observation, workflowId) { + const kinds = {'/api/worker/workflow-tasks/poll': 'workflow_task', '/api/worker/activity-tasks/poll': 'activity_task'}; + assert.equal(item.method, 'POST'); + assert.ok(Object.hasOwn(kinds, item.path), 'only workflow/activity long-poll capacity is tolerated'); + assert.equal(item.client_cancelled ?? false, false); + assert.equal(item.transport_error ?? null, null); + const response = item.response; + assert.ok(response && typeof response === 'object' && !Array.isArray(response), 'actual capacity response'); + assert.equal(response.task, null, 'capacity refusal leases no task'); + assert.equal(response.poll_status, 'long_poll_capacity_exhausted'); + assert.equal(response.reason, 'long_poll_capacity_exhausted'); + assert.equal(response.retryable, true); + assert.equal(response.task_kind, kinds[item.path]); + assert.equal(response.namespace, observation.namespace); + assert.equal(response.task_queue, observation.task_queue); + assert.ok(Number.isSafeInteger(response.retry_after_seconds) && response.retry_after_seconds > 0 + && response.retry_after_seconds <= 25, 'bounded retry hint'); + assert.equal(item.response_retry_after, String(response.retry_after_seconds), 'Retry-After header agrees with the body'); + assert.equal(item.request.worker_id, `${workflowId}:${phase.phase}`); + assert.equal(item.request.task_queue, observation.task_queue); + assert.ok(typeof item.request.poll_request_id === 'string' && item.request.poll_request_id.length > 0); + const later = phase.requests.slice(index + 1); + const ok = request => request.status >= 200 && request.status < 300; + const retry = later.find(request => request.method === item.method && request.path === item.path + && request.request?.poll_request_id === item.request.poll_request_id && ok(request)); + if (retry) { + assert.deepEqual(retry.request, item.request, 'retry changes no poll authority or request fields'); + return; + } + assert.ok(later.some(request => request.method === 'DELETE' && request.path.startsWith('/api/worker/registrations/') && ok(request)), + 'the same poll succeeds later, or the worker stops cleanly after the refusal'); +} + function checkPhpPollPressure(item, index, phase, observation, workflowId) { assert.equal(item.method, 'POST'); assert.equal(item.path, '/api/worker/workflow-tasks/poll', 'PHP pressure is confined to workflow polling'); @@ -264,6 +302,10 @@ export function checkPatchDeployment(fixture, observation, workflowId) { assert.deepEqual(retry.request, item.request, 'retry changes no poll authority or request fields'); continue; } + if (item.status === 429) { + checkLongPollCapacity(item, index, phase, observation, workflowId); + continue; + } assert.ok(item.status >= 200 && item.status < 300, 'all other observed I/O succeeds'); } } diff --git a/tests/Unit/ServerParityPublishedPatchContractTest.mjs b/tests/Unit/ServerParityPublishedPatchContractTest.mjs index a76fe908c..ff09f9f1f 100644 --- a/tests/Unit/ServerParityPublishedPatchContractTest.mjs +++ b/tests/Unit/ServerParityPublishedPatchContractTest.mjs @@ -111,6 +111,48 @@ for (const language of ['rust', 'python']) for (const checkpoint of ['pending', const raw = pressureModel(); mutate(raw.patch_deployment.replacement.requests); assert.throws(() => checkPatchDeployment(fixture, raw, 'test')); }); + function capacityModel(path = '/api/worker/activity-tasks/poll', retried = false) { + const raw = model(); + const request = {worker_id: 'test:replacement', task_queue: 'server-parity-v1', poll_request_id: 'capacity-poll', timeout_seconds: 1}; + const refusal = {method: 'POST', path, status: 429, request, response_retry_after: '1', client_cancelled: false, transport_error: null, + response: {task: null, poll_status: 'long_poll_capacity_exhausted', reason: 'long_poll_capacity_exhausted', + error: 'Worker poll wait capacity is temporarily exhausted.', message: 'Retry the poll after the advertised delay.', + namespace: 'default', task_queue: 'server-parity-v1', task_kind: path.includes('activity') ? 'activity_task' : 'workflow_task', + wait_pool: 'worker', retry_after_seconds: 1, retryable: true}}; + const requests = raw.patch_deployment.replacement.requests; + requests.splice(requests.length - 1, 0, refusal, + ...(retried ? [{...structuredClone(refusal), status: 200, response_retry_after: null, response: {task: null}}] : [])); + return raw; + } + test(`${language} ${checkpoint}: model accepts long-poll capacity before clean shutdown`, () => + checkPatchDeployment(fixture, capacityModel(), 'test')); + test(`${language} ${checkpoint}: model accepts long-poll capacity with same-request retry`, () => + checkPatchDeployment(fixture, capacityModel('/api/worker/workflow-tasks/poll', true), 'test')); + for (const [name, mutate] of [ + ['hidden leased task', requests => requests.at(-2).response.task = {task_id: 'unknown-task'}], + ['other poll status', requests => requests.at(-2).response.poll_status = 'rejected'], + ['contradicting reason', requests => requests.at(-2).response.reason = 'backend_lock_pressure'], + ['not retryable', requests => requests.at(-2).response.retryable = false], + ['wrong task kind', requests => requests.at(-2).response.task_kind = 'workflow_task'], + ['other queue', requests => requests.at(-2).response.task_queue = 'other'], + ['missing Retry-After', requests => requests.at(-2).response_retry_after = null], + ['Retry-After disagrees', requests => requests.at(-2).response_retry_after = '2'], + ['oversized retry hint', requests => { requests.at(-2).response.retry_after_seconds = 26; requests.at(-2).response_retry_after = '26'; }], + ['query poll', requests => requests.at(-2).path = '/api/worker/query-tasks/poll'], + ['completion refused', requests => requests.at(-2).path = '/api/worker/activity-tasks/task/complete'], + ['other worker', requests => requests.at(-2).request.worker_id = 'other-worker'], + ['missing poll identity', requests => delete requests.at(-2).request.poll_request_id], + ['neither retried nor stopped', requests => requests.push(requests.splice(-2, 1)[0])], + ]) test(`${language} ${checkpoint}: capacity model rejects ${name}`, () => { + const raw = capacityModel(); mutate(raw.patch_deployment.replacement.requests); + assert.throws(() => checkPatchDeployment(fixture, raw, 'test')); + }); + test(`${language} ${checkpoint}: capacity model rejects a changed retry request`, () => { + const raw = capacityModel('/api/worker/workflow-tasks/poll', true); + raw.patch_deployment.replacement.requests.at(-2).request.timeout_seconds = 2; + raw.patch_deployment.replacement.requests.at(-2).request.poll_request_id = 'capacity-poll'; + assert.throws(() => checkPatchDeployment(fixture, raw, 'test')); + }); function cancelledModel() { const raw = model(); const requests = raw.patch_deployment.replacement.requests; diff --git a/tests/Unit/Support/composer-sinks.json b/tests/Unit/Support/composer-sinks.json index 6380f4d2c..2207dfa5d 100644 --- a/tests/Unit/Support/composer-sinks.json +++ b/tests/Unit/Support/composer-sinks.json @@ -313,14 +313,14 @@ }, { "path": "scripts/ci/run-composer.sh", - "line": "-v \"$ca:/checked-ca:ro\" --entrypoint composer \"$image\" \"$@\" || status=$?", + "line": "-v \"$ca:/checked-ca:ro\" --entrypoint composer \"$image\" \"$@\" 2>&1 1>&3 3>&- | tee \"$errors\" >&2 3>&-; } 3>&1", "count": 1, "confinement": "The image comes from brotli-safe-composer.sh, which refuses it unless brotli-libs is at least 1.2.1-r0 and owns every Brotli decoder PHP links. The same command sets COMPOSER_CAFILE to the checked store and COMPOSER_CAPATH to an empty directory after the caller's options, which may not name either or an entrypoint.", "guards": [ { "type": "brotli-floor", "line": "image=$(sh \"$here/brotli-safe-composer.sh\")", - "distance": 9 + "distance": 20 } ] }, diff --git a/tests/Unit/Support/composer_trust_test.py b/tests/Unit/Support/composer_trust_test.py index 7727117de..a52353639 100644 --- a/tests/Unit/Support/composer_trust_test.py +++ b/tests/Unit/Support/composer_trust_test.py @@ -378,6 +378,49 @@ def test_the_ca_pair_and_entrypoint_follow_the_callers_options(self) -> None: self.assertRegex(text, r'image=\$\(sh "\$here/brotli-safe-composer\.sh"\)') self.assertRegex(text, r'sh "\$here/checked-ca\.sh" "\$ca"') + def retrying(self, failures: int, message: str, *arguments: str) -> tuple[subprocess.CompletedProcess[str], int]: + """Runs a copy of the wrapper whose Composer run fails `failures` times with `message`.""" + work = Path(tempfile.mkdtemp(prefix="dw-run-composer-retry-")) + self.addCleanup(shutil.rmtree, work, ignore_errors=True) + (work / "bin").mkdir() + shutil.copy(WRAPPER_SCRIPT, work / "run-composer.sh") + (work / "brotli-safe-composer.sh").write_text("#!/bin/sh\necho composer:checked\n") + (work / "checked-ca.sh").write_text('#!/bin/sh\nmkdir -p "$1/no-ca-directory"\n: > "$1/ca.pem"\n') + count = work / "runs" + docker = work / "bin" / "docker" + docker.write_text(f"""#!/bin/sh +n=$(($(cat '{count}' 2>/dev/null || echo 0) + 1)); echo "$n" > '{count}' +if [ "$n" -le {failures} ]; then echo '{message}' >&2; exit 100; fi +echo installed +""") + (work / "bin" / "sleep").write_text("#!/bin/sh\nexit 0\n") + for stub in (docker, work / "bin" / "sleep"): + stub.chmod(0o755) + result = subprocess.run(["bash", str(work / "run-composer.sh"), *arguments], capture_output=True, text=True, + env={**os.environ, "PATH": f"{work / 'bin'}:{os.environ['PATH']}"}, timeout=60) + return result, int(count.read_text()) + + def test_a_locked_install_retries_transient_download_failures(self) -> None: + rate_limited = ('Failed to download a/b from dist: The "https://codeload.github.com/a/b/legacy.zip/c" ' + "file could not be downloaded (HTTP/2 429 )") + result, runs = self.retrying(2, rate_limited, "--rm", "--", "install", "--no-dev") + self.assertEqual((0, 3), (result.returncode, runs), result.stderr) + self.assertEqual("installed\n", result.stdout) + self.assertEqual(2, result.stderr.count("HTTP/2 429"), "each attempt's stderr reaches the caller") + + result, runs = self.retrying(3, rate_limited, "--rm", "--", "install") + self.assertEqual((100, 3), (result.returncode, runs), "three attempts at most, then Composer's own status") + + def test_other_failures_commands_and_kept_containers_are_not_retried(self) -> None: + rate_limited = 'The "https://codeload.github.com/a" file could not be downloaded (HTTP/2 429 )' + for failure, arguments in ( + ("Your lock file does not contain a compatible set of packages.", ("--rm", "--", "install")), + (rate_limited, ("--rm", "--", "update")), + (rate_limited, ("--name", "kept-sdk", "--", "install"))): + with self.subTest(arguments): + result, runs = self.retrying(1, failure, *arguments) + self.assertEqual((100, 1), (result.returncode, runs), result.stderr) + def test_the_floor_refuses_the_pinned_image_and_accepts_its_own(self) -> None: docker_or_skip(self) refused = subprocess.run(["sh", str(FLOOR_SCRIPT), "--check", PINNED_COMPOSER],