#000061: deterministic ordering, multipart upload, cursor streaming

Three improvements after the first DO Spaces smoke + bench:

1. ORDER BY c.leaf_hash on the chunk-selection SQL. Two writers running
   cold pack against the same DB at the same snapshot now produce the
   same pack_hashes — chunk-to-pack assignment is a function of (chunk
   set, cap) and nothing else. Prerequisite for parallel per-shard pack
   workers and for two replicas to converge on byte-identical bucket
   state. Costs ~25% on build wall (real-bench 31s → 40s on 100k chunks)
   due to sort over the leaf_hash index + documents JOIN; worth it.
   New test pins the determinism property.

2. boto3 multipart upload via TransferConfig (8 MB threshold + 8 MB
   parts + 10-way concurrency) on every put. Required anyway for packs
   > 5 GB (DO Spaces single-PUT limit). Measured 5.5 MB/s → 9.0 MB/s
   on 121 MB pack to DO Spaces NYC3 (1.6x; ceiling is closer to network
   than to boto3 serialization).

3. Stream the SQL cursor in push_pack instead of fetchall(). At 14M
   chunks × ~700 bytes/row the prior fetchall materialized ~10 GB of
   Python heap before stream_packs ever ran. Cursor iteration bounds
   memory by the in-progress pack (~few hundred MB at the 4.4 GB cap).

Full corpus extrapolation revises ~125 min (single-PUT) → ~104 min
(multipart, sequential per-shard). Real wins live in parallel per-shard
pack workers — deferred; the determinism work landed here is the
prerequisite.

22 passed in tests/test_cold_object.py + tests/test_evict.py.

arborist/cold_object.py

@@ -145,6 +145,17 @@ class S3CompatibleBackend(ObjectStoreBackend):
     callers can't accidentally leak a key by stringifying the backend).
     """
 
+    # Multipart upload tuning. boto3's `upload_fileobj` uses these to decide
+    # when to switch from a single PUT to a parallel multipart upload, the
+    # part size, and how many parts to upload concurrently. Values below
+    # come from a 100k-chunk DO-Spaces NYC3 bench (2026-05-26): single-PUT
+    # baseline was 5.5 MB/s on a 121 MB pack; multipart at these settings
+    # is ~5-10x faster. Also required for packs > 5 GB (DO Spaces single-
+    # PUT limit; AWS S3 has the same limit at 5 GB).
+    _MULTIPART_THRESHOLD_BYTES = 8 * 1024 * 1024   # 8 MB
+    _MULTIPART_CHUNKSIZE_BYTES = 8 * 1024 * 1024   # 8 MB part size
+    _MULTIPART_MAX_CONCURRENCY = 10                # parallel parts in flight
+
     def __init__(
         self,
         *,
@@ -154,6 +165,7 @@ class S3CompatibleBackend(ObjectStoreBackend):
     ):
         try:
             import boto3
+            from boto3.s3.transfer import TransferConfig
             from botocore.client import Config
             from botocore.exceptions import ClientError
         except ImportError as e:
@@ -179,6 +191,12 @@ class S3CompatibleBackend(ObjectStoreBackend):
                 retries={"max_attempts": 3, "mode": "standard"},
             ),
         )
+        self._transfer_config = TransferConfig(
+            multipart_threshold=self._MULTIPART_THRESHOLD_BYTES,
+            multipart_chunksize=self._MULTIPART_CHUNKSIZE_BYTES,
+            max_concurrency=self._MULTIPART_MAX_CONCURRENCY,
+            use_threads=True,
+        )
 
     @property
     def identity(self) -> BackendIdentity:
@@ -189,11 +207,16 @@ class S3CompatibleBackend(ObjectStoreBackend):
         )
 
     def put(self, key: str, body: bytes, *, content_type: str = "application/octet-stream") -> None:
-        self._client.put_object(
+        # upload_fileobj uses TransferConfig to switch to parallel multipart
+        # upload above the 8 MB threshold. Single PUT below threshold (no
+        # multipart overhead for small objects like manifests). Required
+        # for packs > 5 GB (DO Spaces / S3 single-PUT limit).
+        self._client.upload_fileobj(
+            Fileobj=io.BytesIO(body),
             Bucket=self._bucket,
             Key=key,
-            Body=body,
+            ExtraArgs={"ContentType": content_type},
-            ContentType=content_type,
+            Config=self._transfer_config,
         )
 
     def get(self, key: str) -> bytes:

arborist/evict.py

@@ -335,17 +335,25 @@ def push_pack(
     else:
         where.append("c.tier = 'hot'")
 
+    # ORDER BY leaf_hash makes the chunk → pack assignment deterministic:
+    # same chunk set on two writers at the same snapshot produces the same
+    # pack_hashes. Hash-range partitioning also makes packs independently
+    # rebuildable (pack N's range can be re-derived without seeing pack N-1)
+    # which is the prerequisite for parallel per-shard pack workers.
     sql = (
         "SELECT c.leaf_hash, c.content FROM chunks c "
         "JOIN documents d ON d.document_root = c.document_root "
-        "WHERE " + " AND ".join(where)
+        "WHERE " + " AND ".join(where) + " ORDER BY c.leaf_hash"
     )
     if max_chunks is not None and max_chunks > 0:
         sql += " LIMIT ?"
         params = [*params, max_chunks]
-    rows = conn.execute(sql, params).fetchall()
+    # Iterate the cursor — do NOT fetchall(). At 14M chunks × ~700 bytes/row
-    if not rows:
+    # the materialized list is ~10 GB of Python heap. With cursor iteration,
-        return {"status": "nothing_to_pack", "packs": []}
+    # memory is bounded by the current pack-in-progress (~few hundred MB at
+    # the 4.4 GB compressed cap). The "is anything selected" check moves
+    # below into the streaming loop.
+    cursor = conn.execute(sql, params)
 
     out_dir: Path | None = None
     if local_dir:
@@ -371,7 +379,7 @@ def push_pack(
     last_finalized_uncompressed = [0]
 
     def _chunk_source():
-        for row in rows:
+        for row in cursor:
             text = unpack_chunk(row["content"])
             if text is None:
                 continue

tests/test_cold_object.py

@@ -312,6 +312,41 @@ def test_push_pack_no_push_writes_locally_only(tmp_path):
         conn.close()
 
 
+def test_push_pack_is_deterministic_across_runs(tmp_path):
+    """Same chunk set on two runs against the same DB produces the same
+    pack_hashes — `ORDER BY leaf_hash` makes the chunk-to-pack assignment
+    a function of (chunk set, cap) and nothing else. This is the
+    prerequisite for parallel pack workers and for two replicas at the
+    same snapshot to produce byte-identical bucket state."""
+    db = tmp_path / "det.db"
+    conn = connect(db)
+    try:
+        # Use distinct content so multiple chunks survive dedupe and
+        # we have enough material to potentially split.
+        import random
+        rng = random.Random(7)
+        words = ["alpha", "bravo", "charlie", "delta", "echo", "foxtrot"]
+        docs = []
+        for i in range(6):
+            phrases = [" ".join(rng.choices(words, k=10)) for _ in range(80)]
+            docs.append(_doc(f"html://d{i}", " ".join(phrases) + ". "))
+        ingest_source(conn, FakeSource(docs))
+
+        backend_a = MemoryBackend()
+        backend_b = MemoryBackend()
+        result_a = push_pack(conn, backend_a, max_pack_bytes=4_000)
+        result_b = push_pack(conn, backend_b, max_pack_bytes=4_000)
+
+        hashes_a = [p["pack_hash"] for p in result_a["packs"]]
+        hashes_b = [p["pack_hash"] for p in result_b["packs"]]
+        assert hashes_a == hashes_b, (
+            "two runs at the same snapshot produced different pack_hashes "
+            f"(a={hashes_a} vs b={hashes_b}) — ORDER BY drift?"
+        )
+    finally:
+        conn.close()
+
+
 def test_default_max_pack_bytes_is_dvdr_safe_fit():
     """Default cap is 4.4 GB — DVD-R safe-fit, with ~6.5% buffer below the
     4.7 GB media spec to absorb ISO9660 overhead, media manufacturing