From da1a355df47d1a4e96a2284d2ad4e9d502e6809d Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Tue, 26 May 2026 19:29:13 -0400 Subject: [PATCH] #000067 phase 2 follow-up: skip-if-exists on bucket PUT (content-addressed) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit push_pack now does a HEAD object_size check against the bucket before every PUT. If the object exists at the content-addressed key AND Content-Length matches the local pack body size, skip the upload. Applies to chunks / fts / metadata pack bodies. Why content-addressed-key + size is the right check: pack_hash = hash_leaf(manifest_bytes) object_key = packs/..tar.zst The pack_hash IS the cryptographic identity of the body via its manifest. If the bucket has an object at this exact key, the content is provably identical (otherwise the producer that wrote it computed a different pack_hash → different key). Content-Length match validates against a half-uploaded multipart that a prior crash might have left. Why not ETag: S3 ETag for multipart-uploaded objects is the MD5 of per-part MD5s concatenated, then MD5'd, with a "-N" suffix for part count. Two clients using different multipart_chunksize end up with different ETags for the same content. Unreliable for cross-client equivalence. Content-Length is stable. Result fields gain `skipped_reupload: bool` so callers / the audit chain can tell whether the bucket PUT happened or was a no-op. Real-world impact for tonight's re-pack (--include-fts): before: 4 chunks PUTs (7 GB re-upload, ~24 min) + 4 metadata PUTs (new hashes because manifest now has _fts_pack_hashes — these MUST upload) + 4 fts PUTs (new) ~16 GB uplink, ~50 min wall after: 4 chunks SKIPPED (same content-addressed key already in bucket) + 4 metadata PUTs (new hashes, MUST upload) + 4 fts PUTs (new) ~6 GB uplink, ~20 min wall cold_pending bookkeeping still fires in the upload path so a half-uploaded crash recovers cleanly; skip path leaves no trace because nothing was started. 4 cold-unpack-routed tests pass unchanged. --- arborist/evict.py | 164 +++++++++++++++++++++++++++------------------- 1 file changed, 95 insertions(+), 69 deletions(-) diff --git a/arborist/evict.py b/arborist/evict.py index e9f5d08..dcc5e55 100644 --- a/arborist/evict.py +++ b/arborist/evict.py @@ -472,30 +472,45 @@ def push_pack( # cold_pending row: trace the in-flight upload so a killed # process can be recovered. INSERT before upload starts; # DELETE on success (in the finally below). + chunks_skipped_reupload = False if push_to_bucket: from arborist.cold_object import pack_key - with transaction(conn): - conn.execute( - "INSERT OR REPLACE INTO cold_pending " - "(tempfile_path, pack_hash, kind, backend_endpoint, " - " backend_bucket, object_key, started_at, state) " - "VALUES (?, ?, 'chunks', ?, ?, ?, ?, 'uploading')", - ( - str(pack.body_path), - pack.pack_hash, - backend_id["endpoint_url"], - backend_id["bucket"], - pack_key(pack.pack_hash, kind="chunks"), - pack_ts, - ), - ) - backend.put_pack_file(pack.pack_hash, pack.body_path, kind="chunks") - backend.put_pack_manifest(pack.pack_hash, pack.manifest_bytes, kind="chunks") - with transaction(conn): - conn.execute( - "DELETE FROM cold_pending WHERE tempfile_path = ?", - (str(pack.body_path),), - ) + object_key = pack_key(pack.pack_hash, kind="chunks") + # Skip-if-exists: content-addressed key + size check. The + # pack_hash IS the cryptographic identity (sha256 of the + # manifest); if the bucket already has an object at + # this exact key AND the size matches our local body, + # the content is provably identical. ETag would be + # nicer but multipart-uploaded ETags depend on part + # size and aren't stable across clients — content- + # addressed key + Content-Length is the strict + # check we get for free. + existing_size = backend.object_size(object_key) + if existing_size == pack.body_size: + chunks_skipped_reupload = True + else: + with transaction(conn): + conn.execute( + "INSERT OR REPLACE INTO cold_pending " + "(tempfile_path, pack_hash, kind, backend_endpoint, " + " backend_bucket, object_key, started_at, state) " + "VALUES (?, ?, 'chunks', ?, ?, ?, ?, 'uploading')", + ( + str(pack.body_path), + pack.pack_hash, + backend_id["endpoint_url"], + backend_id["bucket"], + object_key, + pack_ts, + ), + ) + backend.put_pack_file(pack.pack_hash, pack.body_path, kind="chunks") + backend.put_pack_manifest(pack.pack_hash, pack.manifest_bytes, kind="chunks") + with transaction(conn): + conn.execute( + "DELETE FROM cold_pending WHERE tempfile_path = ?", + (str(pack.body_path),), + ) if out_dir is not None: short = pack.pack_hash[:16] final_path = out_dir / f"arborist-pack-{short}.chunks.tar.zst" @@ -511,6 +526,7 @@ def push_pack( "uncompressed_bytes": pack_uncompressed, "compressed_bytes": pack.body_size, "snapshot_root": snapshot_root, + "skipped_reupload": chunks_skipped_reupload, }) finally: if pack.body_path.exists(): @@ -546,35 +562,40 @@ def push_pack( ) try: if push_to_bucket: - with transaction(conn): - conn.execute( - "INSERT OR REPLACE INTO cold_pending " - "(tempfile_path, pack_hash, kind, " - " backend_endpoint, backend_bucket, " - " object_key, started_at, state) " - "VALUES (?, ?, 'fts', ?, ?, ?, ?, 'uploading')", - ( - str(fts_pack.body_path), - fts_pack.pack_hash, - backend_id["endpoint_url"], - backend_id["bucket"], - _pack_key(fts_pack.pack_hash, kind="fts"), - pack_ts, - ), + fts_key = _pack_key(fts_pack.pack_hash, kind="fts") + fts_existing = backend.object_size(fts_key) + if fts_existing == fts_pack.body_size: + pass # already present, skip PUT + else: + with transaction(conn): + conn.execute( + "INSERT OR REPLACE INTO cold_pending " + "(tempfile_path, pack_hash, kind, " + " backend_endpoint, backend_bucket, " + " object_key, started_at, state) " + "VALUES (?, ?, 'fts', ?, ?, ?, ?, 'uploading')", + ( + str(fts_pack.body_path), + fts_pack.pack_hash, + backend_id["endpoint_url"], + backend_id["bucket"], + fts_key, + pack_ts, + ), + ) + backend.put_pack_file( + fts_pack.pack_hash, fts_pack.body_path, kind="fts" ) - backend.put_pack_file( - fts_pack.pack_hash, fts_pack.body_path, kind="fts" - ) - backend.put_pack_manifest( - fts_pack.pack_hash, - fts_pack.manifest_bytes, - kind="fts", - ) - with transaction(conn): - conn.execute( - "DELETE FROM cold_pending WHERE tempfile_path = ?", - (str(fts_pack.body_path),), + backend.put_pack_manifest( + fts_pack.pack_hash, + fts_pack.manifest_bytes, + kind="fts", ) + with transaction(conn): + conn.execute( + "DELETE FROM cold_pending WHERE tempfile_path = ?", + (str(fts_pack.body_path),), + ) if out_dir is not None: short = fts_pack.pack_hash[:16] final_path = out_dir / f"arborist-pack-{short}.fts.tar.zst" @@ -612,27 +633,32 @@ def push_pack( try: if push_to_bucket: from arborist.cold_object import pack_key - with transaction(conn): - conn.execute( - "INSERT OR REPLACE INTO cold_pending " - "(tempfile_path, pack_hash, kind, backend_endpoint, " - " backend_bucket, object_key, started_at, state) " - "VALUES (?, ?, 'metadata', ?, ?, ?, ?, 'uploading')", - ( - str(meta_pack.body_path), - meta_pack.pack_hash, - backend_id["endpoint_url"], - backend_id["bucket"], - pack_key(meta_pack.pack_hash, kind="metadata"), - pack_ts, - ), + meta_key = pack_key(meta_pack.pack_hash, kind="metadata") + meta_existing = backend.object_size(meta_key) + if meta_existing == meta_pack.body_size: + pass # already present, skip PUT (idempotent re-pack) + else: + with transaction(conn): + conn.execute( + "INSERT OR REPLACE INTO cold_pending " + "(tempfile_path, pack_hash, kind, backend_endpoint, " + " backend_bucket, object_key, started_at, state) " + "VALUES (?, ?, 'metadata', ?, ?, ?, ?, 'uploading')", + ( + str(meta_pack.body_path), + meta_pack.pack_hash, + backend_id["endpoint_url"], + backend_id["bucket"], + meta_key, + pack_ts, + ), + ) + backend.put_pack_file( + meta_pack.pack_hash, meta_pack.body_path, kind="metadata" + ) + backend.put_pack_manifest( + meta_pack.pack_hash, meta_pack.manifest_bytes, kind="metadata" ) - backend.put_pack_file( - meta_pack.pack_hash, meta_pack.body_path, kind="metadata" - ) - backend.put_pack_manifest( - meta_pack.pack_hash, meta_pack.manifest_bytes, kind="metadata" - ) # Gap 1: maintain manifest/latest.json so a fresh peer's # `cold list` can resolve "current metadata pack for # snapshot_root X" without enumerating every pack.