engineering/arborist
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

#000067 phase 2 follow-up: skip-if-exists on bucket PUT (content-addressed)

Browse source 
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/<pack_hash>.<kind>.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.
This commit is contained in:
russell@unturf.com 2026-05-26 19:29:13 -04:00
parent 0554199070
commit da1a355df4
No known key found for this signature in database
1 changed files with 95 additions and 69 deletions
Download patch file Download diff file Expand all files Collapse all files

164
arborist/evict.py
View file

@ -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.