From 074524362524c638d558724b4857cd23c58f4f79 Mon Sep 17 00:00:00 2001 From: artin Date: Thu, 13 Aug 2026 15:12:48 +0800 Subject: [PATCH] fix: parallelize Kuikly release publication Signed-off-by: artin --- .github/workflows/task93-kuikly-release.yml | 4 +- scripts/kuikly_maven_publish.py | 42 +++++++++++-------- scripts/test_kuikly_release_contract.py | 45 ++++++++++++++++++++- 3 files changed, 70 insertions(+), 21 deletions(-) diff --git a/.github/workflows/task93-kuikly-release.yml b/.github/workflows/task93-kuikly-release.yml index e5860cbed..aae5009a9 100644 --- a/.github/workflows/task93-kuikly-release.yml +++ b/.github/workflows/task93-kuikly-release.yml @@ -404,7 +404,7 @@ jobs: }} needs: [contract, assemble-candidate, normal-linux, normal-macos, ios-renderer, ohos-gradle, ohos-renderer] runs-on: ubuntu-latest - timeout-minutes: 45 + timeout-minutes: 90 environment: raft-artifacts-production env: EXPECTED_SOURCE_TREE: ${{ inputs.source_tree }} @@ -550,7 +550,7 @@ jobs: PYTHONDONTWRITEBYTECODE=1 python3 "$control_plane/scripts/kuikly_maven_publish.py" plan \ --manifest "$final/manifest.json" --bundle "$final/bundle" \ --output "$final/public-plan.json" - jq -e '.state == "PARTIAL_EXACT" and .presentCount == 69 and .productFileCount == 920 and + jq -e '.state == "PARTIAL_EXACT" and .presentCount >= 69 and .presentCount < 920 and .productFileCount == 920 and (.unexpected | length) == 0 and (.different | length) == 0 and (.listingDisagreement | length) == 0 and .completionPresent == false' \ "$final/public-plan.json" >/dev/null diff --git a/scripts/kuikly_maven_publish.py b/scripts/kuikly_maven_publish.py index 87e73dd89..e0893e3d3 100644 --- a/scripts/kuikly_maven_publish.py +++ b/scripts/kuikly_maven_publish.py @@ -12,6 +12,7 @@ import argparse import base64 +import concurrent.futures import hashlib import json import os @@ -40,6 +41,7 @@ PUBLISH_USERNAME = "raft-ci" PUBLISH_TOKEN_ENV = "RAFT_ARTIFACTS_PUBLISH_TOKEN" USER_AGENT = "kuikly-maven-publish/1.0" +MAX_PARALLEL_REQUESTS = 16 class PublishError(RuntimeError): @@ -210,10 +212,8 @@ def classify(http: Http, manifest: dict[str, Any], bundle: Path) -> dict[str, An remote_owned = {key for key in listing if any(key.startswith(prefix) for prefix in prefixes)} unexpected = sorted(remote_owned - expected_paths - {MANIFEST_PATH}) - present: list[str] = [] - different: list[str] = [] - listing_disagreement: list[str] = [] - for relative, entry in sorted(expected.items()): + def inspect(item: tuple[str, dict[str, Any]]) -> tuple[str, bool, bool, bool]: + relative, entry = item status, body = public_get(http, relative) require(status in {200, 404}, f"public GET HTTP {status}: {relative}") listed = relative in listing @@ -221,12 +221,15 @@ def classify(http: Http, manifest: dict[str, Any], bundle: Path) -> dict[str, An checksum_unlisted_but_readable = ( checksum_descriptor(relative) is not None and found and not listed ) - if listed != found and not checksum_unlisted_but_readable: - listing_disagreement.append(relative) - if found: - present.append(relative) - if len(body) != entry["size"] or sha256_bytes(body) != entry["sha256"]: - different.append(relative) + disagreement = listed != found and not checksum_unlisted_but_readable + differs = found and (len(body) != entry["size"] or sha256_bytes(body) != entry["sha256"]) + return relative, found, differs, disagreement + + with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_PARALLEL_REQUESTS) as executor: + inspected = list(executor.map(inspect, sorted(expected.items()))) + present = [relative for relative, found, _, _ in inspected if found] + different = [relative for relative, _, differs, _ in inspected if differs] + listing_disagreement = [relative for relative, _, _, disagreement in inspected if disagreement] manifest_status, manifest_body = public_get(http, MANIFEST_PATH) require(manifest_status in {200, 404}, f"public completion GET HTTP {manifest_status}") @@ -312,14 +315,15 @@ def put_and_readback(http: Http, token: str, entry: dict[str, Any], body: bytes, def verify_products(http: Http, bundle: Path, objects: Sequence[dict[str, Any]]) -> list[dict[str, Any]]: - readback: list[dict[str, Any]] = [] - for entry in objects: + def verify(entry: dict[str, Any]) -> dict[str, Any]: expected = read_bundle(bundle, entry["path"], entry) status, body = public_get(http, entry["path"]) require(status == 200, f"public readback HTTP {status}: {entry['path']}") require(body == expected, f"public readback differs: {entry['path']}") - readback.append(entry) - return readback + return entry + + with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_PARALLEL_REQUESTS) as executor: + return list(executor.map(verify, objects)) def release( @@ -375,11 +379,15 @@ def persist() -> None: persist() try: - for entry in objects: + def publish(entry: dict[str, Any]) -> tuple[dict[str, Any], str]: body = read_bundle(bundle, entry["path"], entry) result = put_and_readback(http, token, entry, body, content_type="application/octet-stream") - receipt["reused" if result.startswith("reused") else "uploaded"].append(entry["path"]) - persist() + return entry, result + + with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_PARALLEL_REQUESTS) as executor: + for entry, result in executor.map(publish, objects): + receipt["reused" if result.startswith("reused") else "uploaded"].append(entry["path"]) + persist() readback = verify_products(http, bundle, objects) require(canonical_set_digest(readback) == manifest["setSha256"], "public product set digest mismatch") diff --git a/scripts/test_kuikly_release_contract.py b/scripts/test_kuikly_release_contract.py index 5adab9306..7eaadc6b4 100644 --- a/scripts/test_kuikly_release_contract.py +++ b/scripts/test_kuikly_release_contract.py @@ -11,6 +11,8 @@ import subprocess import tarfile import tempfile +import threading +import time import unittest import urllib.parse import xml.etree.ElementTree as ET @@ -707,6 +709,11 @@ def test_workflow_locks_bash_no_mutation_and_no_aggregate_publication(self) -> N "the 920-path public preflight must retain enough time for slow anonymous readback", ) publish_workflow = workflow.split("\n publish:\n", 1)[1] + self.assertIn( + " timeout-minutes: 90\n", + publish_workflow, + "the bounded parallel writer must retain a full retry budget", + ) self.assertIn( " candidate_run_id:\n" " description: Terminal publish=false run whose exact producer shards the writer must reuse\n" @@ -756,9 +763,9 @@ def test_workflow_locks_bash_no_mutation_and_no_aggregate_publication(self) -> N publish_workflow, ) self.assertIn( - "'.state == \"PARTIAL_EXACT\" and .presentCount == 69 and .productFileCount == 920", + "'.state == \"PARTIAL_EXACT\" and .presentCount >= 69 and .presentCount < 920 and .productFileCount == 920", publish_workflow, - "this bounded recovery must stop before PUT unless public preflight is exactly 69/920", + "a resumed writer must accept only a non-regressing exact partial publication", ) self.assertEqual( 5, @@ -1898,6 +1905,40 @@ def test_ohos_har_rejects_nonregular_entries(self) -> None: class PublisherTests(unittest.TestCase): + def test_product_publication_is_bounded_and_parallel(self) -> None: + class TracksParallelPuts(MavenStateHttp): + def __init__(self) -> None: + super().__init__() + self.active = 0 + self.peak = 0 + self.lock = threading.Lock() + + def request(self, origin, path, method, **kwargs): + if origin != contract.PUBLIC_MAVEN_ORIGIN or method != "PUT": + return super().request(origin, path, method, **kwargs) + with self.lock: + self.active += 1 + self.peak = max(self.peak, self.active) + try: + time.sleep(0.01) + return super().request(origin, path, method, **kwargs) + finally: + with self.lock: + self.active -= 1 + + with tempfile.TemporaryDirectory() as raw: + root = Path(raw) + manifest, bundle_root, _ = assemble_fixture(root) + http = TracksParallelPuts() + plan = publisher.classify(http, manifest, bundle_root) + publisher.release( + http, manifest, plan, bundle_root, MavenStateHttp.TOKEN, + root / "execution.json", + ) + + self.assertGreater(http.peak, 1) + self.assertLessEqual(http.peak, publisher.MAX_PARALLEL_REQUESTS) + 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))