diff --git a/arborist/cold_object.py b/arborist/cold_object.py index 71a4402..b1c135d 100644 --- a/arborist/cold_object.py +++ b/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, - ContentType=content_type, + ExtraArgs={"ContentType": content_type}, + Config=self._transfer_config, ) def get(self, key: str) -> bytes: diff --git a/arborist/evict.py b/arborist/evict.py index 7576c4f..b0d072c 100644 --- a/arborist/evict.py +++ b/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() - if not rows: - return {"status": "nothing_to_pack", "packs": []} + # Iterate the cursor — do NOT fetchall(). At 14M chunks × ~700 bytes/row + # the materialized list is ~10 GB of Python heap. With cursor iteration, + # 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 diff --git a/tests/test_cold_object.py b/tests/test_cold_object.py index 21619b3..16875c6 100644 --- a/tests/test_cold_object.py +++ b/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