Shared fixtures: accept published long-poll capacity refusals and retry transient Composer downloads

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"

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');
       }
     }

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;

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
         }
       ]
     },

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],