bundle.index background job: object writes, fetch readiness, and web-scale correctness

The one principle

Content-addressing is a proof, not a label. sha256(bytes) == object_id proves the content. The bundle hash sha256(wire_bytes) == bundle_key proves every byte inside the bundle — including every object's content and its declared ID. Any step that re-verifies what the bundle hash already proved is checking the work of the hash. Delete it.

Current state

process_bundle_index_job exists and the tests pass. It does:

• Fetches bundle from MinIO by bundle_key
• Unpacks msgpack
• Reconstructs snapshot manifests from delta chain → bulk INSERT musehub_snapshots
• Topological sort + bulk INSERT musehub_commits

What it does not do:

• Write object blobs to MinIO under their sha256 keys
• INSERT rows into musehub_objects
• Upsert object refs (path → object_id per snapshot)

The consequence: after a bundle push + index job, commits and snapshots are in PG but individual objects are not addressable by object_id. Fetch, clone, and pull cannot serve file content. The stream push ghost guard fails on any subsequent push because musehub_objects has no rows for this repo.

This is the root cause of the "1 snapshot object(s) not registered" error on stream push after bundle push.

Why per-object sha256 re-verification must NOT be added

The sync path already verified sha256(wire_bytes) == bundle_key. That proof authenticated every bit of wire_bytes, including every object's content bytes and their declared IDs. Re-hashing each object in the background job is checking the work of the bundle hash. Do not add it. Trust the math.

Phase 1 — Object writes: MinIO + PG (the missing step)

This is the load-bearing fix. Everything downstream (fetch, clone, pull) is blocked on it.

What to add to process_bundle_index_job

After unpacking, extract raw_objects = bundle.get("objects") or [].

Step A — parallel MinIO writes

Objects in the bundle may be zstd-compressed (encoding="zstd"). Decompress before writing to MinIO so the stored bytes are the raw content that sha256 was computed over.

import zstandard as _zstd
_dctx = _zstd.ZstdDecompressor()

async def _put_one(obj: dict) -> str:
    oid = obj["object_id"]
    raw = obj["content"]
    if obj.get("encoding") == "zstd":
        raw = _dctx.decompress(raw)
    await backend.put_object(oid, raw)
    return oid

Cap concurrency with _get_blob_put_sem() (100 slots, already exists). Chunk into batches of 200 and asyncio.gather within each chunk. No per-object sha256 check — the bundle hash already proved every byte.

Step B — bulk INSERT musehub_objects

_OBJ_BATCH = 500
obj_rows = [
    {
        "object_id":   obj["object_id"],
        "size_bytes":  len(raw_content),   # decompressed size
        "storage_uri": backend.uri_for(obj["object_id"]),
        "content_cache": None,
    }
    for obj in raw_objects
]
# chunked pg_insert ... on_conflict_do_nothing

Step C — upsert object refs

For each resolved snapshot manifest (already computed in the delta reconstruction loop), upsert one row per (repo_id, object_id) into musehub_object_refs so the ghost guard and fetch presign queries can find them.

Use the existing _upsert_object_refs helper if present, otherwise a chunked pg_insert ... on_conflict_do_nothing.

Web scale math

Serial is a non-starter at web scale. Parallel writes are mandatory.

Acceptance criteria

• After job runs: SELECT COUNT(*) FROM musehub_objects WHERE ... == objects in bundle
• After job runs: a subsequent stream push does not fail the ghost guard
• After job runs: individual objects are retrievable by object_id via the fetch path
• No per-object sha256 verification anywhere in the job

Phase 2 — Fetch readiness: serve objects before the worker runs

The problem: there is a window between unpack-bundle returning 200 and the bundle.index job completing where a clone/fetch would fail — the branch points to a commit, but the objects that commit references are not yet in MinIO.

The fix: content_cache

Store object bytes inline in a content_cache column on musehub_objects at INSERT time, with storage_uri = 'pending'. The fetch path checks content_cache first; if non-null, serves from there. The background job (or a separate promotion job) then PUTs to MinIO and clears content_cache.

This means:

• Sync unpack-bundle path: INSERT objects with content_cache = raw_bytes, storage_uri = 'pending'
• Background job: for each object where storage_uri = 'pending': PUT to MinIO, set storage_uri = real URI, set content_cache = NULL
• Fetch path: COALESCE(content_cache, MinIO GET)

The tradeoff: PG row size. A 25 KB blob inline in PG costs ~25 KB of table storage. For 5k objects that is 125 MB in PG. Acceptable for a push window; not acceptable permanently. Hence the two-step: inline for immediacy, promoted to MinIO for permanence.

Size guard: objects above a configurable threshold (default 1 MB) skip content_cache and go straight to MinIO. Only small objects benefit from inline serving.

Acceptance criteria

• Immediately after unpack-bundle returns 200 (before worker runs): fetch of any file in the pushed commit returns the correct bytes
• After worker runs: content_cache is NULL for all objects, storage_uri points to MinIO
• Objects > 1 MB never enter content_cache

Phase 3 — Idempotency and retry safety

The job must be safe to run multiple times. Network failures, worker crashes, and OOM kills must not corrupt state.

Invariants:

• All PG inserts use on_conflict_do_nothing — already true for snapshots and commits, must be true for objects and object refs
• MinIO PUT is idempotent by nature (same key, same bytes — S3 accepts it)
• Job status must not be set to done until all three write phases complete (objects, snapshots, commits). The worker already wraps in try/except and calls fail_job on error — verify this is wired correctly for partial failures mid-job.
• A job that fails mid-object-write will be retried by stale claim recovery (_STALE_CLAIM_MINUTES). On retry, already-written objects are skipped via on_conflict_do_nothing, already-written snapshots and commits are skipped the same way.

Acceptance criteria

• Running the job twice produces the same DB state as running it once
• Killing the worker mid-job and restarting results in a completed correct state
• No orphan content_cache rows after successful promotion

Phase 4 — Observability

The job is the critical path for fetch/clone correctness. It needs enough logging to diagnose failures in production.

Timing breakdown log (already partially present — extend):

[bundle.index] job=sha256:abc123 repo=sha256:def456
  fetch_bundle:    0.312s  (47,231 KB)
  unpack:          0.841s  (1,035 commits, 5,165 objects, 1,035 snapshots)
  object_puts:     0.287s  (5,165 objects, 100 parallel)
  object_insert:   0.043s  (5,165 rows)
  object_refs:     0.118s  (362,550 refs)
  snapshot_insert: 0.031s  (1,035 rows)
  commit_insert:   0.029s  (1,035 rows)
  total:           1.661s

Job result dict — extend return value:

{
    "commits_written":   1035,
    "snapshots_written": 1035,
    "objects_written":   5165,
    "object_refs_written": 362550,
    "elapsed_ms":        1661,
}

Implementation order

Phase 1 first — it unblocks fetch/clone/pull and fixes the ghost guard race. Everything else is hardening.

Phase 2 — implement only after Phase 1 is proven. It adds complexity (two-phase object storage) and should be gated on a measured fetch-before-worker test showing actual latency.

Phase 3 — implement alongside Phase 1 (idempotency is zero extra cost when you write Phase 1 correctly the first time).

Phase 4 — extend the existing log lines while writing Phase 1.

Files touched

TDD test plan

All tests in musehub/tests/test_bundle_index_job.py.

test_objects_written_to_minio_after_job
  After job runs: backend.get_object(oid) returns correct bytes for every object in bundle.

test_musehub_objects_rows_after_job
  After job runs: musehub_objects has one row per object with correct storage_uri.

test_object_refs_after_job
  After job runs: musehub_object_refs has rows for every (repo_id, object_id) in bundle.

test_no_per_object_sha256_verification
  Patch hashlib.sha256 to raise inside the job — job must complete without calling it.
  (bundle hash already proved content; re-hashing is checking the work of the hash)

test_ghost_guard_passes_after_job
  stream push immediately after bundle.index job completes does not fail ghost guard.

test_job_is_idempotent
  Run process_bundle_index_job twice with same job_id — DB state identical to one run.

test_job_timing_breakdown_in_return_value
  Return dict contains objects_written, object_refs_written, elapsed_ms.