diff --git a/s3proxy/crypto.py b/s3proxy/crypto.py
index 05213c4..b58ca10 100644
--- a/s3proxy/crypto.py
+++ b/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)
diff --git a/s3proxy/handlers/buckets.py b/s3proxy/handlers/buckets.py
index 78f19ef..4961a30 100644
--- a/s3proxy/handlers/buckets.py
+++ b/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:
diff --git a/s3proxy/handlers/multipart/upload_part.py b/s3proxy/handlers/multipart/upload_part.py
index c49623f..cfc3048 100644
--- a/s3proxy/handlers/multipart/upload_part.py
+++ b/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,
diff --git a/s3proxy/routing/dispatcher.py b/s3proxy/routing/dispatcher.py
index 8a49254..f05613a 100644
--- a/s3proxy/routing/dispatcher.py
+++ b/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
diff --git a/s3proxy/state/manager.py b/s3proxy/state/manager.py
index 981b9da..995ebf5 100644
--- a/s3proxy/state/manager.py
+++ b/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,
diff --git a/s3proxy/state/models.py b/s3proxy/state/models.py
index d9a6d0d..99292f6 100644
--- a/s3proxy/state/models.py
+++ b/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""
diff --git a/s3proxy/state/serialization.py b/s3proxy/state/serialization.py
index 9857852..e383d40 100644
--- a/s3proxy/state/serialization.py
+++ 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(
diff --git a/s3proxy/xml_responses.py b/s3proxy/xml_responses.py
index a71eaa3..0d4f89c 100644
--- a/s3proxy/xml_responses.py
+++ b/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"
diff --git a/tests/integration/test_sequential_part_numbering.py b/tests/integration/test_sequential_part_numbering.py
index 08106de..9789106 100644
--- a/tests/integration/test_sequential_part_numbering.py
+++ b/tests/integration/test_sequential_part_numbering.py
@@ -188,6 +188,108 @@ async def test_eight_part_upload_all_sequential(self, manager, settings):
         # Verify Part 8 is small but valid (last part)
         assert state.parts[8].plaintext_size == 356864  # 0.34MB, last part OK
 
+    @pytest.mark.asyncio
+    async def test_clickhouse_600_part_upload_dense_internal_numbers(self, manager, settings):
+        """Regression: sparse range allocation fails at client part 501 (internal 10001).
+
+        ClickHouse shadow backups commonly exceed 500 x 5MB parts per tar.
+        Dense single-internal mode maps client part N → internal N.
+        """
+        bucket = "clickhouse-backups"
+        key = "huge-backup.tar"
+        upload_id = "test-600-part"
+
+        dek = crypto.generate_dek()
+        await manager.create_upload(bucket, key, upload_id, dek, kid="AKIAIOSFODNN7EXAMPLE")
+
+        for part_num in range(1, 601):
+            start = await manager.allocate_internal_parts(
+                bucket, key, upload_id, 1, client_part_number=part_num
+            )
+            assert start == part_num, (
+                f"part {part_num} should map to internal {part_num}, got {start}"
+            )
+
+        state = await manager.get_upload(bucket, key, upload_id)
+        assert state.dense_single_internal is True
+
+    @pytest.mark.asyncio
+    async def test_legacy_sparse_state_inferred_not_dense(self, manager, settings):
+        """Pre-dense Redis state with sparse singles must not switch to dense mid-upload."""
+        bucket = "legacy"
+        key = "backup.tar"
+        upload_id = "legacy-sparse"
+        dek = crypto.generate_dek()
+        await manager.create_upload(bucket, key, upload_id, dek, kid="AKIAIOSFODNN7EXAMPLE")
+
+        # Simulate parts 1-2 already uploaded under old sparse mapping (client 2 → 21).
+        from s3proxy.state import InternalPartMetadata, PartMetadata
+
+        await manager.add_part(
+            bucket,
+            key,
+            upload_id,
+            PartMetadata(
+                part_number=1,
+                plaintext_size=5 * 1024 * 1024,
+                ciphertext_size=5 * 1024 * 1024,
+                etag="e1",
+                internal_parts=[InternalPartMetadata(1, 5 * 1024 * 1024, 5 * 1024 * 1024, "i1")],
+            ),
+        )
+        await manager.add_part(
+            bucket,
+            key,
+            upload_id,
+            PartMetadata(
+                part_number=2,
+                plaintext_size=5 * 1024 * 1024,
+                ciphertext_size=5 * 1024 * 1024,
+                etag="e2",
+                internal_parts=[InternalPartMetadata(21, 5 * 1024 * 1024, 5 * 1024 * 1024, "i21")],
+            ),
+        )
+
+        # Strip dense flag as pre-2026.7.18 Redis blobs would have lacked it.
+        from s3proxy.state.serialization import json_dumps, json_loads
+
+        sk = manager._storage_key(bucket, key, upload_id)
+        raw = await manager._store.get(sk)
+        legacy = json_loads(raw)
+        legacy.pop("dense_single_internal", None)
+        await manager._store.set(sk, json_dumps(legacy), manager._ttl)
+
+        state = await manager.get_upload(bucket, key, upload_id)
+        assert state.dense_single_internal is False
+
+        start = await manager.allocate_internal_parts(
+            bucket, key, upload_id, 1, client_part_number=3
+        )
+        assert start == 41
+
+    @pytest.mark.asyncio
+    async def test_scylla_multi_internal_clears_dense_mode(self, manager, settings):
+        """First multi-internal client part switches back to sparse ranges."""
+        bucket = "scylla-backups"
+        key = "big-Data.db"
+        upload_id = "test-scylla"
+
+        dek = crypto.generate_dek()
+        await manager.create_upload(bucket, key, upload_id, dek, kid="AKIAIOSFODNN7EXAMPLE")
+
+        start = await manager.allocate_internal_parts(
+            bucket, key, upload_id, 8, client_part_number=1
+        )
+        assert start == 1
+
+        state = await manager.get_upload(bucket, key, upload_id)
+        assert state.dense_single_internal is False
+
+        start2 = await manager.allocate_internal_parts(
+            bucket, key, upload_id, 1, client_part_number=2
+        )
+        assert start2 == 21
+
     @pytest.mark.asyncio
     async def test_old_behavior_with_buffer_would_fail(self, manager, settings):
         """
diff --git a/tests/unit/test_dispatch_v1_list.py b/tests/unit/test_dispatch_v1_list.py
index 4130bc0..ecfc44f 100644
--- a/tests/unit/test_dispatch_v1_list.py
+++ b/tests/unit/test_dispatch_v1_list.py
@@ -41,6 +41,7 @@ def __init__(self):
     handle_list_objects = _record("v2")
     forward_request = _record("forward")
     handle_get_bucket_location = _record("location")
+    handle_get_bucket_versioning = _record("versioning")
     handle_list_buckets = _record("list_buckets")
 
 
@@ -68,7 +69,7 @@ def test_v2_list_still_routes_to_v2():
 
 def test_subresource_get_still_forwards():
     assert _route("GET", "/bkt", "acl") == "forward"
-    assert _route("GET", "/bkt", "versioning") == "forward"
+    assert _route("GET", "/bkt", "versioning") == "versioning"
     assert _route("GET", "/bkt", "location") == "location"
 
 
diff --git a/tests/unit/test_routing.py b/tests/unit/test_routing.py
index f5b9be5..385781f 100644
--- a/tests/unit/test_routing.py
+++ b/tests/unit/test_routing.py
@@ -15,6 +15,7 @@
     QUERY_TAGGING,
     QUERY_UPLOAD_ID,
     QUERY_UPLOADS,
+    QUERY_VERSIONING,
     RequestDispatcher,
 )
 
@@ -249,15 +250,10 @@ def test_edge_cases(self):
 class TestQueryStringRouting:
     """Test routing based on query string parameters."""
 
-    def test_versioning_query_forwarded(self):
-        """Test ?versioning is forwarded to backend."""
+    def test_versioning_query_handled_locally(self):
+        """Test ?versioning is answered locally (backend returns 400)."""
         query = "versioning"
-        # This should not match any special handlers
-        assert QUERY_UPLOADS not in query
-        assert QUERY_UPLOAD_ID not in query
-        assert QUERY_LIST_TYPE not in query
-        assert QUERY_LOCATION not in query
-        assert QUERY_DELETE not in query
+        assert QUERY_VERSIONING in query
 
     def test_acl_query_forwarded(self):
         """Test ?acl is forwarded to backend."""
diff --git a/tests/unit/test_xml_responses.py b/tests/unit/test_xml_responses.py
index 9bd7dca..b255521 100644
--- a/tests/unit/test_xml_responses.py
+++ b/tests/unit/test_xml_responses.py
@@ -151,6 +151,16 @@ def test_ap_region(self):
         assert root.text == "ap-northeast-1"
 
 
+class TestBucketVersioning:
+    """Test GetBucketVersioning XML."""
+
+    def test_suspended_status(self):
+        xml = xml_responses.bucket_versioning("Suspended")
+        root = ET.fromstring(xml)
+        assert root.tag.endswith("VersioningConfiguration")
+        assert root.findtext("{http://s3.amazonaws.com/doc/2006-03-01/}Status") == "Suspended"
+
+
 class TestCopyObjectResult:
     """Test CopyObjectResult XML."""