diff --git a/scripts/kuikly_maven_publish.py b/scripts/kuikly_maven_publish.py index e0893e3d3..0cf26e549 100644 --- a/scripts/kuikly_maven_publish.py +++ b/scripts/kuikly_maven_publish.py @@ -17,6 +17,7 @@ import json import os import sys +import time import urllib.error import urllib.parse import urllib.request @@ -41,7 +42,9 @@ PUBLISH_USERNAME = "raft-ci" PUBLISH_TOKEN_ENV = "RAFT_ARTIFACTS_PUBLISH_TOKEN" USER_AGENT = "kuikly-maven-publish/1.0" -MAX_PARALLEL_REQUESTS = 16 +MAX_PARALLEL_REQUESTS = 8 +PUT_MAX_ATTEMPTS = 5 +PUT_RETRYABLE_STATUS = {423, 429, 500, 502, 503, 504} class PublishError(RuntimeError): @@ -301,17 +304,24 @@ def exact_remote(http: Http, entry: dict[str, Any]) -> bool: def put_and_readback(http: Http, token: str, entry: dict[str, Any], body: bytes, *, content_type: str) -> str: if exact_remote(http, entry): return "reused" - status, _ = http.request( - PUBLIC_MAVEN_ORIGIN, - repository_path(entry["path"]), - "PUT", - body=body, - token=token, - content_type=content_type, - ) - require(status in {200, 201, 204, 409}, f"PUT failed with HTTP {status}: {entry['path']}") - require(exact_remote(http, entry), f"uploaded bytes not publicly readable: {entry['path']}") - return "uploaded" if status != 409 else "reused-after-race" + for attempt in range(1, PUT_MAX_ATTEMPTS + 1): + status, _ = http.request( + PUBLIC_MAVEN_ORIGIN, + repository_path(entry["path"]), + "PUT", + body=body, + token=token, + content_type=content_type, + ) + if status in {200, 201, 204, 409}: + require(exact_remote(http, entry), f"uploaded bytes not publicly readable: {entry['path']}") + return "uploaded" if status != 409 else "reused-after-race" + if status not in PUT_RETRYABLE_STATUS or attempt == PUT_MAX_ATTEMPTS: + raise PublishError(f"PUT failed with HTTP {status}: {entry['path']}") + time.sleep(0.25 * (2 ** (attempt - 1))) + if exact_remote(http, entry): + return "reused-after-transient" + raise AssertionError("unreachable PUT retry state") def verify_products(http: Http, bundle: Path, objects: Sequence[dict[str, Any]]) -> list[dict[str, Any]]: diff --git a/scripts/test_kuikly_release_contract.py b/scripts/test_kuikly_release_contract.py index 7eaadc6b4..0feb68ee1 100644 --- a/scripts/test_kuikly_release_contract.py +++ b/scripts/test_kuikly_release_contract.py @@ -1939,6 +1939,46 @@ def request(self, origin, path, method, **kwargs): self.assertGreater(http.peak, 1) self.assertLessEqual(http.peak, publisher.MAX_PARALLEL_REQUESTS) + def test_transient_put_retries_only_after_exact_readback(self) -> None: + entry = {"path": "object.bin", "sha256": contract.sha256_bytes(b"bytes"), "size": 5} + + class TransientThenSuccess(MavenStateHttp): + def __init__(self) -> None: + super().__init__() + self.puts = 0 + + def request(self, origin, path, method, **kwargs): + if origin == contract.PUBLIC_MAVEN_ORIGIN and method == "PUT": + self.puts += 1 + if self.puts == 1: + return 503, b"" + return super().request(origin, path, method, **kwargs) + + http = TransientThenSuccess() + with mock.patch.object(publisher.time, "sleep") as sleep: + result = publisher.put_and_readback( + http, MavenStateHttp.TOKEN, entry, b"bytes", + content_type="application/octet-stream", + ) + self.assertEqual("uploaded", result) + self.assertEqual(2, http.puts) + sleep.assert_called_once_with(0.25) + + class TransientButCommitted(MavenStateHttp): + def request(self, origin, path, method, **kwargs): + if origin == contract.PUBLIC_MAVEN_ORIGIN and method == "PUT": + self.public[entry["path"]] = kwargs["body"] + return 503, b"" + return super().request(origin, path, method, **kwargs) + + committed = TransientButCommitted() + with mock.patch.object(publisher.time, "sleep"): + result = publisher.put_and_readback( + committed, MavenStateHttp.TOKEN, entry, b"bytes", + content_type="application/octet-stream", + ) + self.assertEqual("reused-after-transient", result) + def test_checksum_listing_is_optional_but_exact_get_is_required(self) -> None: with tempfile.TemporaryDirectory() as raw: manifest, bundle_root, bundle = assemble_fixture(Path(raw)) @@ -2117,16 +2157,17 @@ class FailsSecondPut(MavenStateHttp): def request(self, origin, path, method, **kwargs): if origin == contract.PUBLIC_MAVEN_ORIGIN and method == "PUT": self.put_count += 1 - if self.put_count == 2: + if 2 <= self.put_count < 2 + publisher.PUT_MAX_ATTEMPTS: return 500, b"injected" return super().request(origin, path, method, **kwargs) execution_path = root / "interrupted.json" http = FailsSecondPut() - with self.assertRaisesRegex(publisher.PublishError, "PUT failed with HTTP 500"): - publisher.release( - http, manifest, plan, bundle_root, MavenStateHttp.TOKEN, execution_path, - ) + with mock.patch.object(publisher.time, "sleep"): + with self.assertRaisesRegex(publisher.PublishError, "PUT failed with HTTP 500"): + publisher.release( + http, manifest, plan, bundle_root, MavenStateHttp.TOKEN, execution_path, + ) execution = json.loads(execution_path.read_text()) self.assertEqual("incomplete-retryable", execution["state"]) self.assertNotIn(contract.MANIFEST_PATH, http.public)