fix: ClickHouse 500+ part uploads and GetBucketVersioning probe

s3proxy/crypto.py

@@ -170,6 +170,22 @@ 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/buckets.py

@@ -282,6 +282,26 @@ async def handle_get_bucket_location(self, request: Request, creds: S3Credential
             except ClientError as e:
                 self._raise_bucket_error(e, bucket)
 
+    async def handle_get_bucket_versioning(
+        self, request: Request, creds: S3Credentials
+    ) -> Response:
+        """Return versioning status without forwarding to backend.
+
+        Hetzner/object-storage backends reject GetBucketVersioning with HTTP 400.
+        ClickHouse backup agents probe ?versioning= on startup; answering locally
+        avoids noisy 400s while keeping uploads unversioned (Suspended).
+        """
+        bucket = self._parse_bucket(request.url.path)
+        async with self._client(creds) as client:
+            try:
+                await client.head_bucket(bucket)
+            except ClientError as e:
+                self._raise_bucket_error(e, bucket)
+        return Response(
+            content=xml_responses.bucket_versioning("Suspended"),
+            media_type="application/xml",
+        )
+
     async def handle_list_multipart_uploads(
         self, request: Request, creds: S3Credentials
     ) -> Response:

s3proxy/handlers/multipart/upload_part.py

@@ -142,7 +142,8 @@ async def handle_upload_part(self, request: Request, creds: S3Credentials) -> Re
                 upload_path="framed" if use_framed else "buffered",
             )
 
-            # Allocate internal part numbers
+            # Per-client allocation: dense 1:1 for all-5MB uploads (ClickHouse 600+
+            # parts), sparse ranges once a client part needs multiple internals (Scylla).
             internal_part_start = await self.multipart_manager.allocate_internal_parts(
                 bucket,
                 key,

s3proxy/routing/dispatcher.py

@@ -16,6 +16,7 @@
 QUERY_PART_NUMBER = "partNumber"
 QUERY_LIST_TYPE = "list-type"
 QUERY_LOCATION = "location"
+QUERY_VERSIONING = "versioning"
 QUERY_DELETE = "delete"
 QUERY_TAGGING = "tagging"
 
@@ -153,7 +154,16 @@ async def _dispatch_bucket(
         if QUERY_LOCATION in query and method == METHOD_GET:
             return await self.handler.handle_get_bucket_location(request, creds)
 
-        skip_queries = (QUERY_LIST_TYPE, QUERY_DELETE, QUERY_UPLOADS, QUERY_LOCATION)
+        if QUERY_VERSIONING in query and method == METHOD_GET:
+            return await self.handler.handle_get_bucket_versioning(request, creds)
+
+        skip_queries = (
+            QUERY_LIST_TYPE,
+            QUERY_DELETE,
+            QUERY_UPLOADS,
+            QUERY_LOCATION,
+            QUERY_VERSIONING,
+        )
         if query and not any(q in query for q in skip_queries):
             # A GET whose query is only listing params is ListObjects V1 (it lacks
             # list-type=2), not a bucket sub-resource. Fall through to the list

s3proxy/state/manager.py

@@ -3,7 +3,12 @@
 import structlog
 from structlog.stdlib import BoundLogger
 
-from ..crypto import MAX_INTERNAL_PARTS_PER_CLIENT, validate_internal_part_allocation
+from ..crypto import (
+    MAX_INTERNAL_PARTS_PER_CLIENT,
+    S3_MAX_PART_NUMBER,
+    allocate_client_internal_range,
+    internal_part_range,
+)
 from ..errors import S3Error
 from .models import (
     MultipartUploadState,
@@ -231,43 +236,109 @@ async def allocate_internal_parts(
         internal part numbers to avoid conflicts.
         """
         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
+                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(
+                        "DENSE_SINGLE_INTERNAL_DISABLED",
+                        bucket=bucket,
+                        key=key,
+                        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(
+                        client_part_number,
+                        count,
+                        dense_single_internal=state.dense_single_internal,
+                    )
+                except ValueError 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,
+                        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"
+                )
+
+                return serialize_upload_state(state)
+
             try:
-                start, end = validate_internal_part_allocation(client_part_number, count)
+                result = await self._store.update(sk, updater, self._ttl)
             except ValueError as e:
-                logger.error(
-                    "INTERNAL_PART_ALLOCATION_REJECTED",
-                    bucket=bucket,
-                    key=key,
-                    client_part=client_part_number,
-                    requested=count,
-                    max_per_client=MAX_INTERNAL_PARTS_PER_CLIENT,
-                    error=str(e),
-                )
                 raise S3Error.invalid_part(str(e)) from e
-
-            if count > MAX_INTERNAL_PARTS_PER_CLIENT:
-                logger.warning(
-                    "INTERNAL_PARTS_EXCEED_RANGE",
-                    bucket=bucket,
-                    key=key,
-                    client_part=client_part_number,
-                    requested=count,
-                    max=MAX_INTERNAL_PARTS_PER_CLIENT,
-                )
+            except StateMissingError as e:
+                raise S3Error.invalid_part(str(e)) from e
+            if result is None:
+                raise StateMissingError(f"Upload state missing for {bucket}/{key}/{upload_id}")
 
             logger.info(
                 "ALLOCATE_INTERNAL_PARTS",
                 bucket=bucket,
                 key=key,
+                upload_id=self._truncate_id(upload_id),
                 client_part=client_part_number,
                 count=count,
                 start=start,
                 end=end,
+                allocation_mode=allocation_mode,
+                dense_single_internal=dense_at_alloc and count == 1,
             )
+            if client_part_number >= 500:
+                logger.warning(
+                    "HIGH_CLIENT_PART_ALLOCATED",
+                    bucket=bucket,
+                    key=key,
+                    upload_id=self._truncate_id(upload_id),
+                    client_part=client_part_number,
+                    internal_start=start,
+                    allocation_mode=allocation_mode,
+                )
             return start
 
         # Fallback: sequential allocation
-        return await self._allocate_sequential(bucket, key, upload_id, count)
+        return await self._allocate_sequential_checked(bucket, key, upload_id, count)
 
     async def _allocate_sequential(self, bucket: str, key: str, upload_id: str, count: int) -> int:
         """Allocate internal parts sequentially (fallback when no client part)."""
@@ -281,6 +352,12 @@ def updater(data: bytes) -> bytes:
                 return data
 
             start = state.next_internal_part_number
+            end = start + count - 1
+            if end > S3_MAX_PART_NUMBER:
+                raise ValueError(
+                    f"upload needs internal parts {start}-{end} "
+                    f"but S3 allows at most {S3_MAX_PART_NUMBER}"
+                )
             state.next_internal_part_number = start + count
 
             logger.debug(
@@ -305,6 +382,23 @@ def updater(data: bytes) -> bytes:
 
         return start
 
+    async def _allocate_sequential_checked(
+        self, bucket: str, key: str, upload_id: str, count: int
+    ) -> int:
+        """Sequential allocation with S3 part-number ceiling enforcement."""
+        try:
+            return await self._allocate_sequential(bucket, key, upload_id, count)
+        except ValueError as e:
+            logger.error(
+                "INTERNAL_PART_ALLOCATION_REJECTED",
+                bucket=bucket,
+                key=key,
+                upload_id=self._truncate_id(upload_id),
+                requested=count,
+                error=str(e),
+            )
+            raise S3Error.invalid_part(str(e)) from e
+
     async def set_deferred_copy_tail(
         self,
         bucket: str,

s3proxy/state/models.py

@@ -52,6 +52,10 @@ class MultipartUploadState:
     total_plaintext_size: int = 0
     next_internal_part_number: int = 1  # Next S3 part number to use
     kid: str = ""  # Key id that wraps this upload's DEK ("" = default key)
+    # True while every client part uses a single internal part (e.g. 5MB ClickHouse
+    # 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
     # 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

@@ -24,6 +24,20 @@ def json_loads(data: bytes) -> dict:
     return orjson.loads(data)
 
 
+def _infer_dense_single_internal(parts: dict[int, PartMetadata]) -> bool:
+    """Infer allocation mode for Redis state written before dense_single_internal existed."""
+    if not parts:
+        return True
+    for part in parts.values():
+        if len(part.internal_parts) > 1:
+            return False
+        if len(part.internal_parts) == 1:
+            ipn = part.internal_parts[0].internal_part_number
+            if ipn != part.part_number:
+                return False
+    return True
+
+
 def serialize_upload_state(state: MultipartUploadState) -> bytes:
     """Serialize upload state to JSON bytes for Redis."""
     part_numbers = sorted(state.parts.keys())
@@ -40,6 +54,7 @@ def serialize_upload_state(state: MultipartUploadState) -> bytes:
         "deferred_copy_tail": base64.b64encode(state.deferred_copy_tail).decode()
         if state.deferred_copy_tail
         else "",
+        "dense_single_internal": state.dense_single_internal,
         "parts": {
             str(pn): {
                 "part_number": p.part_number,
@@ -127,6 +142,18 @@ def deserialize_upload_state(data: bytes) -> MultipartUploadState | None:
             for pn, p in obj.get("parts", {}).items()
         }
 
+        dense_single_internal = obj.get("dense_single_internal")
+        if dense_single_internal is None:
+            dense_single_internal = _infer_dense_single_internal(parts)
+            logger.info(
+                "LEGACY_DENSE_MODE_INFERRED",
+                bucket=obj.get("bucket"),
+                key=obj.get("key"),
+                upload_id=obj.get("upload_id", "")[:20] + "...",
+                dense_single_internal=dense_single_internal,
+                part_count=len(parts),
+            )
+
         return MultipartUploadState(
             dek=base64.b64decode(obj["dek"]),
             bucket=obj["bucket"],
@@ -140,6 +167,7 @@ def deserialize_upload_state(data: bytes) -> MultipartUploadState | None:
             deferred_copy_tail=base64.b64decode(obj["deferred_copy_tail"])
             if obj.get("deferred_copy_tail")
             else b"",
+            dense_single_internal=dense_single_internal,
         )
     except (KeyError, TypeError, ValueError) as e:
         logger.error(

s3proxy/xml_responses.py

@@ -126,6 +126,14 @@ def location_constraint(location: str | None) -> str:
     return f"{_XML_HEADER}\n<LocationConstraint {_S3_NS}>{escape(location)}</LocationConstraint>"
 
 
+def bucket_versioning(status: str = "Suspended") -> str:
+    """Build VersioningConfiguration XML for GetBucketVersioning."""
+    return f"""{_XML_HEADER}
+<VersioningConfiguration {_S3_NS}>
+    <Status>{escape(status)}</Status>
+</VersioningConfiguration>"""
+
+
 def copy_result(etag: str, last_modified: str, is_part: bool = False) -> str:
     """Build CopyObjectResult or CopyPartResult XML."""
     tag = "CopyPartResult" if is_part else "CopyObjectResult"