Fix InvalidPart on multipart uploads when a smaller part arrives before a larger lower-numbered one

.github/workflows/docker-publish.yml

@@ -32,7 +32,7 @@ jobs:
         uses: docker/setup-buildx-action@v4
 
       - name: Log in to Container Registry
-        uses: docker/login-action@v4.6.0
+        uses: docker/login-action@v4.7.0
         with:
           registry: ghcr.io
           username: ${{ github.repository_owner }}

Makefile

@@ -41,6 +41,7 @@ INTEGRATION_multipart_TESTS = \
 	tests/integration/test_part_ordering.py \
 	tests/integration/test_sequential_part_numbering.py \
 	tests/integration/test_sequential_part_numbering_e2e.py \
+	tests/integration/test_multipart_part_placement_e2e.py \
 	tests/integration/test_large_file_streaming.py \
 	tests/integration/test_entity_too_small_errors.py \
 	tests/integration/test_entity_too_small_fix.py \

dashboard-frontend/package.json

@@ -17,7 +17,7 @@
     "svelte": "^5.1.0",
     "svelte-check": "^4.0.0",
     "typescript": "^5.6.0",
-    "vite": "^8.0.16"
+    "vite": "^8.3.5"
   },
   "dependencies": {
     "uplot": "^1.6.31"

pyproject.toml

@@ -20,20 +20,20 @@ keywords = [
 ]
 requires-python = ">=3.14"
 dependencies = [
-    "fastapi>=0.141.1",
-    "uvicorn[standard]>=0.52.4",
+    "fastapi>=0.146.0",
+    "uvicorn[standard]>=0.54.0",
     "aioboto3>=15.5.0",
     "boto3>=1.40.0",
     "botocore>=1.40.0",
-    "cryptography>=50.0.1",
-    "pydantic>=2.13.5",
-    "pydantic-settings>=2.15.0",
+    "cryptography>=50.0.2",
+    "pydantic>=2.14.1",
+    "pydantic-settings>=2.16.0",
     "structlog>=26.1.0",
     "httpx[http2]>=0.28.1",
     "python-multipart>=0.0.32",
     # Performance optimizations
-    "uvloop>=0.22.0; sys_platform != 'win32'",
-    "orjson>=3.12.0",
+    "uvloop>=0.23.0; sys_platform != 'win32'",
+    "orjson>=3.13.0",
     # Redis for distributed state
     "redis[hiredis]>=8.1.0",
     # Memory monitoring
@@ -48,11 +48,11 @@ dev = [
     "pytest-asyncio>=1.4.0",
     "pytest-cov>=7.1.0",
     "pytest-xdist>=3.8.0",
-    "moto[s3]>=5.2.3",
-    "ruff>=0.16.5",
-    "mypy>=2.3.1",
-    "boto3-stubs[s3]>=1.43.83",
-    "fakeredis>=2.37.1",
+    "moto[s3]>=5.2.4",
+    "ruff>=0.17.1",
+    "mypy>=2.4.0",
+    "boto3-stubs[s3]>=1.43.114",
+    "fakeredis>=2.40.0",
     "requests>=2.34.2",
 ]
 

s3proxy/crypto.py

@@ -170,22 +170,6 @@ def internal_part_range(client_part_number: int, count: int) -> tuple[int, int]:
     return start, start + count - 1
 
 
-def allocate_client_internal_range(
-    client_part_number: int,
-    count: int,
-    *,
-    dense_single_internal: bool,
-) -> tuple[int, int]:
-    """Return (start, end) internal part numbers for one client part."""
-    if dense_single_internal and count == 1:
-        if client_part_number > S3_MAX_PART_NUMBER:
-            raise ValueError(
-                f"client part {client_part_number} exceeds S3 part limit {S3_MAX_PART_NUMBER}"
-            )
-        return client_part_number, client_part_number
-    return validate_internal_part_allocation(client_part_number, count)
-
-
 def validate_internal_part_allocation(client_part_number: int, count: int) -> tuple[int, int]:
     """Return (start, end) or raise if the range exceeds S3's part-number ceiling."""
     start, end = internal_part_range(client_part_number, count)

s3proxy/handlers/multipart/lifecycle.py

@@ -382,7 +382,21 @@ def _build_s3_parts(
         if not s3_parts:
             raise S3Error.invalid_part("No valid parts found")
 
-        s3_parts.sort(key=lambda p: p["PartNumber"])
+        # S3 concatenates parts by number, but reads locate each client part by
+        # summing sizes in client order. Sorting here would hide a mismatch and
+        # produce an object that cannot be decrypted.
+        numbers = [p["PartNumber"] for p in s3_parts]
+        if any(a >= b for a, b in zip(numbers, numbers[1:], strict=False)):
+            logger.error(
+                "INTERNAL_PART_ORDER_VIOLATION",
+                bucket=bucket,
+                key=key,
+                upload_id=upload_id[:20] + "...",
+                internal_parts=numbers,
+            )
+            raise S3Error.invalid_part(
+                "Internal part numbers overlap or are out of client-part order; restart the upload"
+            )
         return s3_parts, completed_parts, total_plaintext
 
     async def _complete_multipart_upload_with_retry(

s3proxy/state/allocation.py

@@ -0,0 +1,90 @@
+"""Placement of client parts in an upload's internal S3 part-number space.
+
+S3 assembles the object in ascending part-number order and reads assume the
+ciphertext of client part N starts after that of every lower client part. So
+the internal ranges of an upload must be disjoint and ordered like the client
+parts, however the parts arrive. A part is placed in the free window between
+its already-placed neighbours; the window is also kept large enough for every
+unplaced part between them to get at least one number.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+
+from ..crypto import MAX_INTERNAL_PARTS_PER_CLIENT, S3_MAX_PART_NUMBER, internal_part_range
+
+# While every part so far fits one internal part, the first part to arrive may
+# be the short last part of an upload whose other parts need several. Reserve
+# sparse slots for the parts below it so those can still be placed. The unused
+# numbers are lost for good, so this is capped at 19 per part below it
+# (at most 285 of S3's 10000) to leave room for very large single-internal uploads.
+AMBIGUOUS_FIRST_PART_MAX = 16
+
+
+class PartPlacementError(ValueError):
+    """No placement keeps this client part ordered and disjoint from the others."""
+
+
+def place_client_part(
+    allocations: Mapping[int, tuple[int, int]],
+    client_part: int,
+    count: int,
+    *,
+    single_internal: bool,
+) -> tuple[int, int]:
+    """Return the inclusive internal range for ``client_part`` needing ``count`` parts.
+
+    ``allocations`` maps already-placed client parts to their ranges.
+    ``single_internal`` is True while every part so far fits one internal part:
+    placement is then dense (internal N for client N), as 600-part uploads need
+    to stay under S3's 10000-part limit. Otherwise it is the sparse layout
+    (20 numbers per client part). Either preference is dropped for a compact
+    placement when it would collide; if no placement is safe, raise.
+    """
+    if not 1 <= client_part <= S3_MAX_PART_NUMBER:
+        raise PartPlacementError(f"client part {client_part} is outside 1-{S3_MAX_PART_NUMBER}")
+    if not 1 <= count <= MAX_INTERNAL_PARTS_PER_CLIENT:
+        raise PartPlacementError(
+            f"client part {client_part} needs {count} internal parts "
+            f"but between 1 and {MAX_INTERNAL_PARTS_PER_CLIENT} are allowed"
+        )
+
+    below = above = 0
+    widest = count
+    for part, (placed_start, placed_end) in allocations.items():
+        if part == client_part:
+            continue
+        widest = max(widest, placed_end - placed_start + 1)
+        if part < client_part:
+            below = max(below, part)
+        elif not above or part < above:
+            above = part
+    has_others = len(allocations) > (client_part in allocations)
+    low = allocations[below][1] + 1 if below else 1
+    high = allocations[above][0] - 1 if above else S3_MAX_PART_NUMBER
+    gap_below = client_part - below - 1
+    gap_above = above - client_part - 1 if above else 0
+    first = low + gap_below
+    last = high - count + 1 - gap_above
+    if first > last:
+        raise PartPlacementError(
+            f"client part {client_part} needs {count} internal part(s) but only "
+            f"{high - low + 1 - gap_below - gap_above} are free between the parts already "
+            "placed around it; upload parts of uniform size, or restart the upload"
+        )
+
+    previous = allocations.get(client_part)
+    if previous is not None and first <= previous[0] <= last:
+        return previous[0], previous[0] + count - 1
+
+    if single_internal:
+        ambiguous = not has_others and 1 < client_part <= AMBIGUOUS_FIRST_PART_MAX
+        preferred = internal_part_range(client_part, 1)[0] if ambiguous else client_part
+    else:
+        preferred = internal_part_range(client_part, 1)[0]
+    if first <= preferred <= last:
+        start = preferred
+    else:
+        start = min(max(low + gap_below * widest, first), last)
+    return start, start + count - 1

s3proxy/state/manager.py

@@ -6,10 +6,10 @@
 from ..crypto import (
     MAX_INTERNAL_PARTS_PER_CLIENT,
     S3_MAX_PART_NUMBER,
-    allocate_client_internal_range,
     internal_part_range,
 )
 from ..errors import S3Error
+from .allocation import PartPlacementError, place_client_part
 from .models import (
     MultipartUploadState,
     PartMetadata,
@@ -230,26 +230,25 @@ async def allocate_internal_parts(
         count: int,
         client_part_number: int = 0,
     ) -> int:
-        """Allocate internal part numbers based on client part number.
+        """Allocate internal part numbers for a client part.
 
-        Each client part gets a reserved range of MAX_INTERNAL_PARTS_PER_CLIENT
-        internal part numbers to avoid conflicts.
+        With a client part number, the range is placed by ``place_client_part`` so
+        it stays disjoint from, and ordered like, every other client part's range
+        whatever order they arrive in; the allocation is recorded in the upload
+        state. Without one (UploadPartCopy), numbers are handed out sequentially.
         """
         if client_part_number > 0:
             sk = self._storage_key(bucket, key, upload_id)
             start = 1
             end = 1
             allocation_mode = "sparse"
-            dense_at_alloc = True
 
             def updater(data: bytes) -> bytes:
-                nonlocal start, end, allocation_mode, dense_at_alloc
+                nonlocal start, end, allocation_mode
                 state = deserialize_upload_state(data)
                 if state is None:
                     raise StateMissingError(f"Upload state corrupted for {bucket}/{key}")
 
-                sparse_start, sparse_end = internal_part_range(client_part_number, count)
-                dense_at_alloc = state.dense_single_internal
                 if count > 1 and state.dense_single_internal:
                     state.dense_single_internal = False
                     logger.info(
@@ -259,49 +258,35 @@ def updater(data: bytes) -> bytes:
                         upload_id=self._truncate_id(upload_id),
                         client_part=client_part_number,
                         requested=count,
-                        sparse_range=f"{sparse_start}-{sparse_end}",
                     )
 
                 try:
-                    start, end = allocate_client_internal_range(
+                    start, end = place_client_part(
+                        state.allocations,
                         client_part_number,
                         count,
-                        dense_single_internal=state.dense_single_internal,
+                        single_internal=state.dense_single_internal,
                     )
-                except ValueError as e:
+                except PartPlacementError as e:
                     logger.error(
                         "INTERNAL_PART_ALLOCATION_REJECTED",
                         bucket=bucket,
                         key=key,
                         upload_id=self._truncate_id(upload_id),
                         client_part=client_part_number,
                         requested=count,
-                        dense_single_internal=dense_at_alloc,
-                        sparse_range=f"{sparse_start}-{sparse_end}",
-                        dense_range=f"{client_part_number}-{client_part_number}"
-                        if count == 1
-                        else None,
-                        max_per_client=MAX_INTERNAL_PARTS_PER_CLIENT,
+                        placed_parts=len(state.allocations),
                         error=str(e),
                     )
                     raise
 
-                if count > MAX_INTERNAL_PARTS_PER_CLIENT and not state.dense_single_internal:
-                    logger.warning(
-                        "INTERNAL_PARTS_EXCEED_RANGE",
-                        bucket=bucket,
-                        key=key,
-                        client_part=client_part_number,
-                        requested=count,
-                        max=MAX_INTERNAL_PARTS_PER_CLIENT,
-                    )
-
-                allocation_mode = (
-                    "dense"
-                    if dense_at_alloc and count == 1 and start == client_part_number
-                    else "sparse"
-                )
-
+                state.allocations[client_part_number] = (start, end)
+                if start == client_part_number and count == 1:
+                    allocation_mode = "dense"
+                elif start == internal_part_range(client_part_number, 1)[0]:
+                    allocation_mode = "sparse"
+                else:
+                    allocation_mode = "compact"
                 return serialize_upload_state(state)
 
             try:
@@ -323,7 +308,6 @@ def updater(data: bytes) -> bytes:
                 start=start,
                 end=end,
                 allocation_mode=allocation_mode,
-                dense_single_internal=dense_at_alloc and count == 1,
             )
             if client_part_number >= 500:
                 logger.warning(

s3proxy/state/models.py

@@ -56,6 +56,9 @@ class MultipartUploadState:
     # shadow tars). Maps client part N → internal N so 600-part uploads stay under
     # S3's 10k part limit. Cleared on the first multi-internal client part (Scylla).
     dense_single_internal: bool = True
+    # Inclusive internal part range reserved for each client part, recorded when
+    # the part starts uploading so concurrent and out-of-order parts see it.
+    allocations: dict[int, tuple[int, int]] = field(default_factory=dict)
     # Hybrid passthrough may leave a sub-5MB plaintext suffix here when the
     # client part range ends mid internal frame but more client parts follow.
     deferred_copy_tail: bytes = b""

s3proxy/state/serialization.py

@@ -55,6 +55,7 @@ def serialize_upload_state(state: MultipartUploadState) -> bytes:
         if state.deferred_copy_tail
         else "",
         "dense_single_internal": state.dense_single_internal,
+        "allocations": {str(pn): list(rng) for pn, rng in state.allocations.items()},
         "parts": {
             str(pn): {
                 "part_number": p.part_number,
@@ -154,6 +155,12 @@ def deserialize_upload_state(data: bytes) -> MultipartUploadState | None:
                 part_count=len(parts),
             )
 
+        allocations = {int(pn): (rng[0], rng[1]) for pn, rng in obj.get("allocations", {}).items()}
+        for pn, part in parts.items():
+            if pn not in allocations and part.internal_parts:
+                numbers = [ip.internal_part_number for ip in part.internal_parts]
+                allocations[pn] = (min(numbers), max(numbers))
+
         return MultipartUploadState(
             dek=base64.b64decode(obj["dek"]),
             bucket=obj["bucket"],
@@ -168,6 +175,7 @@ def deserialize_upload_state(data: bytes) -> MultipartUploadState | None:
             if obj.get("deferred_copy_tail")
             else b"",
             dense_single_internal=dense_single_internal,
+            allocations=allocations,
         )
     except (KeyError, TypeError, ValueError) as e:
         logger.error(