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.