perf(static-sites): parallelize Bunny transfers

README.md

@@ -62,6 +62,10 @@ python3 static_site.py bunny dist --dry-run
 python3 static_site.py bunny dist
 ```
 
+Changed files are transferred with 16 workers by default while preserving the
+asset-before-HTML and manifest-last publication order. Set
+`BUNNY_TRANSFER_WORKERS` or pass `--workers` to tune the limit.
+
 Set `BUNNY_STORAGE_ENDPOINT` when the Storage Zone is outside Frankfurt.
 
 ## Development

action.yml

@@ -21,6 +21,10 @@ inputs:
     description: Report changes without uploading or deleting files.
     required: false
     default: "false"
+  workers:
+    description: Maximum concurrent Bunny Storage transfers.
+    required: false
+    default: "16"
 
 runs:
   using: composite
@@ -33,9 +37,10 @@ runs:
         BUNNY_STORAGE_ENDPOINT: ${{ inputs.storage-endpoint }}
         BUNNY_PURGE_ENDPOINT: ${{ inputs.purge-endpoint }}
         STATIC_SITE_DRY_RUN: ${{ inputs.dry-run }}
+        STATIC_SITE_WORKERS: ${{ inputs.workers }}
       run: |
         set -euo pipefail
-        args=(bunny "$STATIC_SITE_DIRECTORY")
+        args=(bunny "$STATIC_SITE_DIRECTORY" --workers "$STATIC_SITE_WORKERS")
         if [[ "$STATIC_SITE_DRY_RUN" == "true" ]]; then
           args+=(--dry-run)
         elif [[ "$STATIC_SITE_DRY_RUN" != "false" ]]; then

static_site.py

@@ -4,6 +4,8 @@
 from __future__ import annotations
 
 import argparse
+from collections.abc import Callable
+from concurrent.futures import ThreadPoolExecutor, as_completed
 import hashlib
 import json
 import mimetypes
@@ -17,19 +19,32 @@ import urllib.parse
 import urllib.request
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any
+from typing import Any, TypeVar
 
 
 MANIFEST_NAME = ".static-site-manifest.json"
 MANIFEST_VERSION = 1
 BUNNY_RETRY_DELAYS = (1, 2, 4, 8, 16)
 BUNNY_TRANSIENT_STATUS_CODES = {401, 408, 429, 500, 502, 503, 504}
+DEFAULT_BUNNY_TRANSFER_WORKERS = 16
+
+T = TypeVar("T")
 
 
 class DeployError(RuntimeError):
     """A static-site deployment could not be completed safely."""
 
 
+def positive_int(value: str) -> int:
+    try:
+        parsed = int(value)
+    except ValueError as exc:
+        raise argparse.ArgumentTypeError("must be an integer") from exc
+    if parsed < 1:
+        raise argparse.ArgumentTypeError("must be at least 1")
+    return parsed
+
+
 @dataclass(frozen=True)
 class LocalFile:
     path: str
@@ -140,6 +155,9 @@ class BunnyStorage:
                     f"Bunny Storage {method} {path} failed: HTTP {exc.code}: {detail}"
                 ) from exc
             except urllib.error.URLError as exc:
+                if attempt < len(BUNNY_RETRY_DELAYS):
+                    time.sleep(BUNNY_RETRY_DELAYS[attempt])
+                    continue
                 raise DeployError(
                     f"Bunny Storage {method} {path} failed: {exc.reason}"
                 ) from exc
@@ -192,6 +210,39 @@ def bunny_plan(
     return upload, delete
 
 
+def run_concurrently(
+    items: list[T], action: Callable[[T], None], workers: int
+) -> None:
+    if workers < 1:
+        raise DeployError("Bunny transfer workers must be at least 1")
+    if not items:
+        return
+    with ThreadPoolExecutor(max_workers=min(workers, len(items))) as executor:
+        futures = [executor.submit(action, item) for item in items]
+        for future in as_completed(futures):
+            future.result()
+
+
+def apply_bunny_plan(
+    storage: BunnyStorage,
+    files: dict[str, LocalFile],
+    upload: list[str],
+    delete: list[str],
+    workers: int = DEFAULT_BUNNY_TRANSFER_WORKERS,
+) -> None:
+    assets = [
+        path for path in upload if not path.lower().endswith((".html", ".htm"))
+    ]
+    documents = [
+        path for path in upload if path.lower().endswith((".html", ".htm"))
+    ]
+
+    run_concurrently(assets, lambda path: storage.upload(files[path]), workers)
+    run_concurrently(documents, lambda path: storage.upload(files[path]), workers)
+    run_concurrently(delete, storage.delete, workers)
+    storage.upload_manifest(manifest_for(files))
+
+
 def purge_bunny_pull_zone(pull_zone_id: str, api_key: str) -> None:
     if not api_key:
         raise DeployError("BUNNY_API_KEY is required when BUNNY_PULL_ZONE_ID is set")
@@ -264,11 +315,7 @@ def deploy_bunny(args: argparse.Namespace) -> dict[str, Any]:
         return result
 
     if result["changed"]:
-        for path in upload:
-            storage.upload(files[path])
-        for path in delete:
-            storage.delete(path)
-        storage.upload_manifest(manifest_for(files))
+        apply_bunny_plan(storage, files, upload, delete, args.workers)
     if args.purge_endpoint:
         purge_bunny_gateway(args.purge_endpoint, args.zone, args.password)
         result["purged"] = True
@@ -335,6 +382,14 @@ def parser() -> argparse.ArgumentParser:
     bunny.add_argument(
         "--purge-endpoint", default=os.environ.get("BUNNY_PURGE_ENDPOINT", "")
     )
+    bunny.add_argument(
+        "--workers",
+        type=positive_int,
+        default=os.environ.get(
+            "BUNNY_TRANSFER_WORKERS", str(DEFAULT_BUNNY_TRANSFER_WORKERS)
+        ),
+        help="Maximum concurrent Bunny Storage transfers.",
+    )
     bunny.add_argument("--dry-run", action="store_true")
     bunny.set_defaults(
         handler=deploy_bunny,

tests/test_static_site.py

@@ -4,6 +4,7 @@ import json
 from pathlib import Path
 import sys
 import tempfile
+import threading
 import unittest
 from unittest import mock
 
@@ -15,11 +16,13 @@ sys.modules[SPEC.name] = static_site
 SPEC.loader.exec_module(static_site)
 
 DeployError = static_site.DeployError
+apply_bunny_plan = static_site.apply_bunny_plan
 bunny_plan = static_site.bunny_plan
 collect_files = static_site.collect_files
 manifest_for = static_site.manifest_for
 parse_manifest = static_site.parse_manifest
 purge_bunny_gateway = static_site.purge_bunny_gateway
+run_concurrently = static_site.run_concurrently
 
 
 def write_site(root: Path) -> None:
@@ -80,6 +83,56 @@ class StaticSiteTests(unittest.TestCase):
             remote = manifest_for(files)["files"]
             self.assertEqual(bunny_plan(files, remote), ([], []))
 
+    def test_bunny_transfers_use_multiple_workers(self) -> None:
+        barrier = threading.Barrier(4)
+        completed: list[int] = []
+
+        def transfer(item: int) -> None:
+            barrier.wait(timeout=2)
+            completed.append(item)
+
+        run_concurrently([1, 2, 3, 4], transfer, workers=4)
+
+        self.assertCountEqual(completed, [1, 2, 3, 4])
+
+    def test_bunny_transfer_order_keeps_documents_and_manifest_safe(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            write_site(root)
+            (root / "about.html").write_text("<h1>About</h1>")
+            files = collect_files(root)
+            events: list[tuple[str, str]] = []
+            storage = mock.Mock()
+            storage.upload.side_effect = lambda item: events.append(
+                ("upload", item.path)
+            )
+            storage.delete.side_effect = lambda path: events.append(("delete", path))
+            storage.upload_manifest.side_effect = lambda _manifest: events.append(
+                ("manifest", static_site.MANIFEST_NAME)
+            )
+
+            apply_bunny_plan(
+                storage,
+                files,
+                ["assets/app.js", "about.html", "index.html"],
+                ["stale.css"],
+                workers=2,
+            )
+
+        asset_position = events.index(("upload", "assets/app.js"))
+        document_positions = {
+            events.index(("upload", "about.html")),
+            events.index(("upload", "index.html")),
+        }
+        delete_position = events.index(("delete", "stale.css"))
+        self.assertTrue(
+            all(asset_position < position for position in document_positions)
+        )
+        self.assertTrue(
+            all(position < delete_position for position in document_positions)
+        )
+        self.assertEqual(events[-1], ("manifest", static_site.MANIFEST_NAME))
+
     def test_gateway_requires_a_plain_https_origin(self) -> None:
         invalid = (
             "http://purge.example",
@@ -115,6 +168,25 @@ class StaticSiteTests(unittest.TestCase):
         sleep.assert_called_once_with(1)
         self.assertEqual(urlopen.call_count, 2)
 
+    @mock.patch.object(static_site.time, "sleep")
+    @mock.patch.object(static_site.urllib.request, "urlopen")
+    def test_storage_retries_transient_connection_errors(
+        self, urlopen: mock.Mock, sleep: mock.Mock
+    ) -> None:
+        response = mock.MagicMock()
+        response.__enter__.return_value.read.return_value = b"manifest"
+        urlopen.side_effect = [
+            static_site.urllib.error.URLError("connection reset"),
+            response,
+        ]
+        storage = static_site.BunnyStorage(
+            "example", "storage-password", "storage.example"
+        )
+
+        self.assertEqual(storage.get_manifest(), b"manifest")
+        sleep.assert_called_once_with(1)
+        self.assertEqual(urlopen.call_count, 2)
+
     @mock.patch.object(static_site.urllib.request, "urlopen")
     def test_gateway_reuses_storage_password(self, urlopen: mock.Mock) -> None:
         urlopen.return_value.__enter__.return_value = object()