fix: retry transient Kuikly publication writes

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

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)