simplify calculating number of parts, raise error when over limit

lib/galaxy/files/sources/invenio.py

@@ -118,7 +118,7 @@ MAX_UPLOAD_PARTS = 10_000
 
 
 def calculate_multipart_params(file_size: int, preferred_part_size: int | None = None) -> tuple[int, int]:
-    """Calculate optimal parts count and part size for multipart upload.
+    """Calculate parts count and part size for multipart upload.
 
     Args:
         file_size: Total file size in bytes
@@ -127,34 +127,34 @@ def calculate_multipart_params(file_size: int, preferred_part_size: int | None =
     Returns:
         Tuple of (parts_count, part_size)
 
+    Raises:
+        ValueError: If the file is larger than MAX_UPLOAD_PARTS * MAX_UPLOAD_PART_SIZE
+            (~48.8 TiB), which exceeds the maximum uploadable size.
+
     Note:
         Maximum uploadable file size is MAX_UPLOAD_PARTS * MAX_UPLOAD_PART_SIZE (~48.8 TiB).
-        Files larger than this will still return valid params but would fail server-side.
+        Files larger than this cannot be uploaded via multipart and raise ValueError.
     """
     if file_size == 0:
         return 1, MIN_UPLOAD_PART_SIZE
 
-    # Start with preferred or minimum part size
+    # Start with preferred or minimum part size, clamped to [min, max]
     part_size = preferred_part_size or MIN_UPLOAD_PART_SIZE
-
-    # Ensure part_size is within bounds
     part_size = max(part_size, MIN_UPLOAD_PART_SIZE)
     part_size = min(part_size, MAX_UPLOAD_PART_SIZE)
 
-    # Calculate parts needed
+    # Grow part_size to the minimum that keeps the part count within MAX_UPLOAD_PARTS.
+    part_size = max(part_size, math.ceil(file_size / MAX_UPLOAD_PARTS))
+    part_size = min(part_size, MAX_UPLOAD_PART_SIZE)
+
+    max_upload_size = MAX_UPLOAD_PARTS * MAX_UPLOAD_PART_SIZE
+    if file_size > max_upload_size:
+        raise ValueError(
+            f"File size {file_size} bytes exceeds the maximum multipart upload size "
+            f"of {max_upload_size} bytes ({MAX_UPLOAD_PARTS} parts x {MAX_UPLOAD_PART_SIZE} bytes)."
+        )
+
     parts = math.ceil(file_size / part_size)
-
-    # If too many parts, increase part size (up to max)
-    while parts > MAX_UPLOAD_PARTS and part_size < MAX_UPLOAD_PART_SIZE:
-        part_size = min(part_size * 2, MAX_UPLOAD_PART_SIZE)
-        parts = math.ceil(file_size / part_size)
-
-    # For extremely large files, cap parts at MAX_UPLOAD_PARTS
-    # This means part_size may effectively be larger than calculated
-    # but such files would likely fail server-side anyway
-    if parts > MAX_UPLOAD_PARTS:
-        parts = MAX_UPLOAD_PARTS
-
     return parts, part_size
 
 

test/unit/files/test_invenio_multipart.py

@@ -60,19 +60,22 @@ class TestCalculateMultipartParams:
         assert part_size >= MIN_UPLOAD_PART_SIZE
 
     def test_calculate_multipart_params_extremely_large_file(self):
-        """Extremely large files should hit both MAX limits.
+        """Files larger than the maximum uploadable size raise ValueError instead of truncating."""
+        # 100 TiB file - exceeds theoretical s3 maximum (~48.8 TiB)
+        file_size = 100 * 1024**4
+        with pytest.raises(ValueError, match="exceeds the maximum multipart upload size"):
+            calculate_multipart_params(file_size)
 
-        Note: Files larger than MAX_UPLOAD_PARTS * MAX_UPLOAD_PART_SIZE (~48.8 TiB)
-        cannot be uploaded via multipart, but we cap params rather than fail here.
-        The upload would fail server-side anyway.
-        """
-        # 100 TiB file - exceeds theoretical maximum (~48.8 TiB)
-        file_size = 100 * 1024**4  # 100 TiB
-        parts, part_size = calculate_multipart_params(file_size)
-        # Parts should be capped at MAX_UPLOAD_PARTS
+    def test_calculate_multipart_params_raises_at_max_boundary(self):
+        """One byte over the maximum uploadable size raises; exactly at the max does not."""
+        max_upload_size = MAX_UPLOAD_PARTS * MAX_UPLOAD_PART_SIZE
+        # Exactly at the boundary is allowed (parts == MAX_UPLOAD_PARTS at MAX part size).
+        parts, part_size = calculate_multipart_params(max_upload_size)
         assert parts == MAX_UPLOAD_PARTS
-        # Part size should hit max
         assert part_size == MAX_UPLOAD_PART_SIZE
+        # One byte over raises.
+        with pytest.raises(ValueError):
+            calculate_multipart_params(max_upload_size + 1)
 
     def test_calculate_multipart_params_respects_preferred_part_size(self):
         """Should use preferred part size when provided and valid."""