Stage 09 · Multipart domain and validation

Goal

Model upload identity, staged parts, completion manifests, size rules, and composite ETags.

Hands-on task

Starting from stage-08, Implement MultipartUpload, StagedPart, receipts, and validate_completion(...). Keep all behavior inside the listed source-like boundaries; do not copy the patch first.

Deliverable files / 交付文件

• src/minis3/errors.py
• src/minis3/multipart.py

Self-check

1. Where is this stage's visibility or state transition owned?

   Answer

   Minimum size is checked at completion because only then is the final part known.

2. Which test would fail first if the new boundary were bypassed?

   Answer

   Read tests.txt, identify the narrowest new node, and name the public call it exercises.

Pass command

uv run pytest -q $(cat journey/stages/09-multipart-domain/tests.txt)

The real S3 lesson

Minimum size is checked at completion because only then is the final part known.

Textbook

Chapter 6

Compare this stage on GitHub

After finishing, use git checkout stage-09 to compare your result.

Try first, then peek: stage.patch

diff --git a/src/minis3/errors.py b/src/minis3/errors.py
index e1a2230..9db3b4c 100644
--- a/src/minis3/errors.py
+++ b/src/minis3/errors.py
@@ -28,3 +28,18 @@ class NoSuchVersion(MiniS3Error):
 class InvalidContinuationToken(MiniS3Error):
     """The list continuation token was malformed or belongs to another query."""

+
+class NoSuchUpload(MiniS3Error):
+    """The addressed multipart upload does not exist or no longer exists."""
+
+
+class InvalidPart(MiniS3Error):
+    """A completion entry names a missing part or the wrong part ETag."""
+
+
+class InvalidPartOrder(MiniS3Error):
+    """Multipart completion entries were not in strictly ascending order."""
+
+
+class EntityTooSmall(MiniS3Error):
+    """A non-final multipart part is below the configured minimum size."""
diff --git a/src/minis3/multipart.py b/src/minis3/multipart.py
new file mode 100644
index 0000000..c10ab02
--- /dev/null
+++ b/src/minis3/multipart.py
@@ -0,0 +1,107 @@
+"""Multipart values and completion validation.
+
+An upload part cannot know whether it will be the final part in the eventual
+completion list. Therefore the S3 minimum-size rule is intentionally checked
+at completion, against every listed part except the last one.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+from hashlib import md5
+
+from .errors import EntityTooSmall, InvalidPart, InvalidPartOrder
+from .model import content_etag
+
+
+MIN_PART_SIZE = 5 * 1024 * 1024
+MAX_PART_NUMBER = 10_000
+
+
+@dataclass(frozen=True, slots=True)
+class MultipartUpload:
+    """Identity of one durable but object-invisible upload."""
+
+    bucket: str
+    key: str
+    upload_id: str
+    sequence: int
+    initiated_at: float
+
+
+@dataclass(frozen=True, slots=True)
+class MultipartPart:
+    """Public receipt returned after one part has been durably staged."""
+
+    part_number: int
+    etag: str
+    size: int
+
+
+@dataclass(frozen=True, slots=True)
+class StagedPart:
+    """Bytes recovered from an upload's private staging directory."""
+
+    part_number: int
+    body: bytes
+
+    @property
+    def etag(self) -> str:
+        return content_etag(self.body)
+
+    @property
+    def size(self) -> int:
+        return len(self.body)
+
+    @property
+    def receipt(self) -> MultipartPart:
+        return MultipartPart(self.part_number, self.etag, self.size)
+
+
+CompletionEntry = MultipartPart | tuple[int, str]
+
+
+def _entry_identity(entry: CompletionEntry) -> tuple[int, str]:
+    if isinstance(entry, MultipartPart):
+        return entry.part_number, entry.etag
+    try:
+        part_number, etag = entry
+    except (TypeError, ValueError) as exc:
+        raise InvalidPart(entry) from exc
+    if not isinstance(part_number, int) or not isinstance(etag, str):
+        raise InvalidPart(entry)
+    return part_number, etag
+
+
+def validate_completion(
+    staged: dict[int, StagedPart],
+    entries: Sequence[CompletionEntry],
+    *,
+    minimum_part_size: int,
+) -> tuple[tuple[StagedPart, ...], str]:
+    """Validate a client manifest and return ordered parts plus composite ETag."""
+
+    identities = tuple(_entry_identity(entry) for entry in entries)
+    if not identities:
+        raise InvalidPart("completion list must contain at least one part")
+    numbers = tuple(part_number for part_number, _etag in identities)
+    if any(left >= right for left, right in zip(numbers, numbers[1:])):
+        raise InvalidPartOrder(numbers)
+
+    selected: list[StagedPart] = []
+    for part_number, expected_etag in identities:
+        part = staged.get(part_number)
+        if part is None or part.etag != expected_etag:
+            raise InvalidPart(f"part {part_number}")
+        selected.append(part)
+    for part in selected[:-1]:
+        if part.size < minimum_part_size:
+            raise EntityTooSmall(f"part {part.part_number}")
+
+    # Multipart ETags hash binary MD5 digests, not their hexadecimal strings.
+    digests = b"".join(
+        md5(part.body, usedforsecurity=False).digest() for part in selected
+    )
+    composite = md5(digests, usedforsecurity=False).hexdigest()
+    return tuple(selected), f'"{composite}-{len(selected)}"'