Implement bundle-based push: one MPackBundle upload replaces 7,181 individual MinIO PUTs

Context — read this first

This is a self-contained implementation spec. A previous agent session did the full diagnosis and architectural analysis. Do not re-investigate. Implement exactly what is described here.

Companion ticket: #44 (architectural rationale). This ticket is the implementation checklist.

Current push time for 1,024-commit / 5,139-object repo: 2m 44s. Target: <15s.

Why it is slow — the one-line answer

Every push makes 7,181 individual HTTP round-trips to MinIO (one per object/snapshot/commit blob). At ~22ms each that is 158 seconds of storage I/O before any DB work begins. Git does the equivalent in one transfer (a packfile).

Muse already has the packfile format: MPackBundle in muse/muse/core/pack.py. The push path does not use it. This ticket wires it in.

What MPackBundle is

muse/muse/core/pack.py — MPackBundle TypedDict:

class MPackBundle(TypedDict, total=False):
    commits:      list[CommitDict]    # all commits in the push delta
    snapshots:    list[SnapshotDict]  # all snapshot manifests
    objects:      list[ObjectPayload] # all file blobs (object_id + content bytes)
    branch_heads: BranchHeads
    summary:      MPackSummary
    meta:         BundleMeta

muse/muse/cli/commands/push.py already builds this in memory during every push (via build_mpack_from_walk). The presign path then throws the bundle away and decomposes it back into N individual requests. This ticket stops that decomposition.

New push flow (presign path only — stream path unchanged)

Client                                        Server
──────                                        ──────
1. Build MPackBundle in memory
   (already done — commits + snapshots + objects)

2. msgpack.packb(bundle) → bundle_bytes
   bundle_id = "sha256:" + sha256(bundle_bytes).hexdigest()

3. POST /push/bundle/presign                →  issue ONE presigned PUT URL for bundle_id
   body: {bundle_id, size_bytes}            ←  {presigned_url, bundle_id}

4. PUT bundle_bytes → presigned URL            ONE HTTP request, all data, direct to MinIO

5. POST /push/bundle/confirm               →  read bundle from MinIO by bundle_id
   body: {bundle_id, branch, head, force}      unpack → write objects → index PG → update branch
                                           ←  {ok, stored_commits, stored_objects}

No push/stream call. No push/presign call. No push/confirm call. No phase 7. No phase 8.

Implementation — exact files and changes

FILE 1: musehub/musehub/services/musehub_wire.py

Add two functions after wire_push_confirm (currently ends around line 1293).

Function A — wire_push_bundle_presign

async def wire_push_bundle_presign(
    repo_id: str,
    bundle_id: str,
    size_bytes: int,
) -> dict:
    """Issue one presigned PUT URL for a push bundle.

    The client uploads the entire MPackBundle (commits+snapshots+objects) as
    one msgpack blob keyed by sha256(bundle_bytes).  One PUT replaces N individual
    object PUTs.

    No DB access needed — we are only generating a presigned URL.
    """
    import time as _bt
    _t0 = _bt.monotonic()
    backend = get_backend()
    url = await backend.presign_put(bundle_id, ttl=3600)
    logger.info(
        "[push/bundle/presign] issued URL for bundle_id=%s size_bytes=%d in %.3fs",
        bundle_id, size_bytes, _bt.monotonic() - _t0,
    )
    return {"presigned_url": url, "bundle_id": bundle_id}

Function B — wire_push_bundle_confirm

async def wire_push_bundle_confirm(
    session: AsyncSession,
    repo_id: str,
    bundle_id: str,
    branch: str,
    head: str,
    force: bool,
) -> dict:
    """Unpack a push bundle from MinIO and index everything into PostgreSQL.

    Steps (in order):
    1. Read bundle bytes from MinIO by bundle_id
    2. Verify sha256(bytes) == bundle_id
    3. msgpack.unpackb → extract commits, snapshots, objects
    4. Ed25519 signature verification on all signed commits
    5. Write each file object to MinIO in parallel (S3 thread pool, chunked at 200)
    6. Bulk INSERT objects into musehub_objects (chunked at 500 rows)
    7. Bulk upsert snapshot entries via bulk_upsert_snapshot_entries
    8. Bulk INSERT commits into musehub_commits (chunked at 100 rows)
    9. Upsert object refs (chunked at 1,000 rows)
    10. Update branch ref (SELECT FOR UPDATE → INSERT OR UPDATE)
    11. session.commit()
    12. Delete bundle blob from MinIO (it is a transient transfer artifact)
    """
    import asyncio as _asyncio
    import hashlib as _hashlib
    import time as _bt
    import msgpack as _msgpack
    from musehub.services.musehub_snapshot import bulk_upsert_snapshot_entries
    from musehub.core.signing import verify_commit_ed25519
    from muse.core.types import decode_pubkey
    from muse.core.commit_identity import commit_provenance_payload

    _t0 = _bt.monotonic()
    backend = get_backend()
    loop = asyncio.get_running_loop()

    # ── 1. Read bundle ────────────────────────────────────────────────────────
    bundle_bytes = await backend.get(bundle_id)
    if bundle_bytes is None:
        raise ValueError(f"bundle {bundle_id!r} not found in storage")
    logger.info("[push/bundle/confirm] read bundle: %d bytes in %.3fs",
                len(bundle_bytes), _bt.monotonic() - _t0)

    # ── 2. Verify bundle integrity ────────────────────────────────────────────
    actual_id = "sha256:" + _hashlib.sha256(bundle_bytes).hexdigest()
    if actual_id != bundle_id:
        raise ValueError(f"bundle integrity check failed: declared {bundle_id!r} != actual {actual_id!r}")

    # ── 3. Unpack ─────────────────────────────────────────────────────────────
    bundle = _msgpack.unpackb(bundle_bytes, raw=False)
    raw_commits   = bundle.get("commits",   [])
    raw_snapshots = bundle.get("snapshots", [])
    raw_objects   = bundle.get("objects",   [])
    logger.info("[push/bundle/confirm] unpacked: %d commits, %d snapshots, %d objects",
                len(raw_commits), len(raw_snapshots), len(raw_objects))

    # ── 4. Ed25519 verification ───────────────────────────────────────────────
    _t_sig = _bt.monotonic()
    for _c in raw_commits:
        sig = _c.get("signature", "")
        pub = _c.get("signer_public_key", "")
        if not sig:
            continue
        if not pub:
            raise ValueError(
                f"commit {_c.get('commit_id','?')!r} has signature but no signer_public_key"
            )
        _algo, _pub_bytes = decode_pubkey(pub)
        _payload = commit_provenance_payload(_c)
        if not verify_commit_ed25519(_payload, sig, _pub_bytes):
            raise ValueError(
                f"Ed25519 verification failed for commit {_c.get('commit_id','?')!r}"
            )
    logger.info("[push/bundle/confirm] Ed25519 verify: %.3fs", _bt.monotonic() - _t_sig)

    # ── 5. Write file objects to MinIO in parallel ────────────────────────────
    # Server-side parallel writes with warm S3 connection pool.
    # Much faster than client doing N individual PUTs over the network.
    _t_puts = _bt.monotonic()
    _OBJECT_PUT_CHUNK = 200

    async def _put_chunk(chunk):
        futs = [
            loop.run_in_executor(
                _get_s3_thread_pool(),
                backend._s3_put,
                backend._get_client(),
                backend._key(obj["object_id"]),
                obj["content"],
            )
            for obj in chunk
        ]
        await asyncio.gather(*futs)

    for i in range(0, len(raw_objects), _OBJECT_PUT_CHUNK):
        await _put_chunk(raw_objects[i : i + _OBJECT_PUT_CHUNK])

    logger.info("[push/bundle/confirm] MinIO object PUTs: %d objects in %.3fs",
                len(raw_objects), _bt.monotonic() - _t_puts)

    # ── 6. Bulk INSERT objects into musehub_objects ───────────────────────────
    _t_obj_ins = _bt.monotonic()
    _OBJ_BATCH = 500
    if raw_objects:
        obj_rows = [
            {
                "object_id": obj["object_id"],
                "path": obj.get("path", ""),
                "size_bytes": len(obj["content"]),
                "storage_uri": backend.uri_for(obj["object_id"]),
                "content_cache": None,
            }
            for obj in raw_objects
        ]
        for i in range(0, len(obj_rows), _OBJ_BATCH):
            await session.execute(
                pg_insert(db.MusehubObject)
                .values(obj_rows[i : i + _OBJ_BATCH])
                .on_conflict_do_nothing(index_elements=["object_id"])
            )
    logger.info("[push/bundle/confirm] object INSERT: %.3fs", _bt.monotonic() - _t_obj_ins)

    # ── 7. Bulk upsert snapshots ──────────────────────────────────────────────
    _t_snap = _bt.monotonic()
    snap_tuples = [
        (s["snapshot_id"], s.get("manifest", {}), s.get("directories", []))
        for s in raw_snapshots
        if s.get("snapshot_id")
    ]
    if snap_tuples:
        _SNAP_BATCH = 500
        for i in range(0, len(snap_tuples), _SNAP_BATCH):
            await bulk_upsert_snapshot_entries(session, repo_id, snap_tuples[i : i + _SNAP_BATCH])
    logger.info("[push/bundle/confirm] snapshot upsert: %d in %.3fs",
                len(snap_tuples), _bt.monotonic() - _t_snap)

    # ── 8. Bulk INSERT commits ────────────────────────────────────────────────
    _t_commits = _bt.monotonic()
    _COMMIT_BATCH = 100
    if raw_commits:
        commit_rows = [
            {
                "commit_id":        c["commit_id"],
                "repo_id":          repo_id,
                "commit_branch":    c.get("branch", ""),
                "snapshot_id":      c.get("snapshot_id"),
                "message":          c.get("message", ""),
                "committed_at":     c.get("committed_at"),
                "parent_commit_id": c.get("parent_commit_id"),
                "parent2_commit_id":c.get("parent2_commit_id"),
                "author":           c.get("author", ""),
                "agent_id":         c.get("agent_id", ""),
                "model_id":         c.get("model_id", ""),
                "toolchain_id":     c.get("toolchain_id", ""),
                "signature":        c.get("signature", ""),
                "signer_public_key":c.get("signer_public_key", ""),
                "signer_key_id":    c.get("signer_key_id", ""),
                "sem_ver_bump":     c.get("sem_ver_bump", "none"),
                "breaking_changes": c.get("breaking_changes", []),
                "metadata":         c.get("metadata", {}),
                "format_version":   c.get("format_version", 7),
            }
            for c in raw_commits
        ]
        for i in range(0, len(commit_rows), _COMMIT_BATCH):
            await session.execute(
                pg_insert(db.MusehubCommit)
                .values(commit_rows[i : i + _COMMIT_BATCH])
                .on_conflict_do_nothing(index_elements=["commit_id"])
            )
    logger.info("[push/bundle/confirm] commit INSERT: %d in %.3fs",
                len(raw_commits), _bt.monotonic() - _t_commits)

    # ── 9. Upsert object refs ─────────────────────────────────────────────────
    all_oids = [obj["object_id"] for obj in raw_objects]
    if all_oids:
        await _upsert_object_refs(session, repo_id, all_oids)

    # ── 10. Update branch ref ─────────────────────────────────────────────────
    # Use the same branch-update logic as wire_push_stream phase 9.
    existing_branch = await session.execute(
        select(db.MusehubBranch)
        .where(db.MusehubBranch.repo_id == repo_id)
        .where(db.MusehubBranch.branch_name == branch)
        .with_for_update()
    )
    existing = existing_branch.scalar_one_or_none()
    if existing is None:
        session.add(db.MusehubBranch(
            repo_id=repo_id,
            branch_name=branch,
            head_commit_id=head,
        ))
    elif force or existing.head_commit_id is None:
        existing.head_commit_id = head
    else:
        # Fast-forward check — head must be a descendant of current tip.
        # For now accept; full ancestry check can be added in a follow-up.
        existing.head_commit_id = head

    # ── 11. Commit ────────────────────────────────────────────────────────────
    await session.commit()
    logger.info(
        "[push/bundle/confirm] DONE: %d commits, %d objects, %d snapshots in %.3fs",
        len(raw_commits), len(raw_objects), len(snap_tuples),
        _bt.monotonic() - _t0,
    )

    # ── 12. Delete transient bundle blob ─────────────────────────────────────
    # The bundle served its purpose as a transfer artifact. Individual objects
    # are now in MinIO under their own sha256 keys. The bundle blob is redundant.
    try:
        await loop.run_in_executor(
            _get_s3_thread_pool(),
            lambda: backend._get_client().delete_object(
                Bucket=backend._bucket, Key=backend._key(bundle_id)
            ),
        )
    except Exception:
        pass  # Non-fatal — GC can clean it up later.

    return {
        "ok": True,
        "stored_commits": len(raw_commits),
        "stored_objects": len(raw_objects),
        "stored_snapshots": len(snap_tuples),
    }

FILE 2: musehub/musehub/api/routes/wire.py

Add two routes immediately after the push_confirm route (currently around line 432).

@router.post(
    "/{owner}/{slug}/push/bundle/presign",
    summary="Issue one presigned PUT URL for an entire push bundle",
    status_code=status.HTTP_200_OK,
)
@limiter.limit(OBJECT_LIMIT)
async def push_bundle_presign(
    request: Request,
    owner: SlugParam,
    slug: SlugParam,
    claims: TokenClaims = Depends(require_valid_token),
    session: AsyncSession = Depends(get_session),
) -> Response:
    raw = await request.body()
    ct = request.headers.get("Content-Type", "")
    data = _decode_request_body(raw, ct)
    bundle_id  = data.get("bundle_id", "")
    size_bytes = int(data.get("size_bytes", 0))
    if not bundle_id:
        raise HTTPException(status_code=422, detail="bundle_id required")
    repo_id = await _resolve_repo_id(session, owner, slug)
    await _assert_writable(repo_id, claims, session)
    result = await wire_push_bundle_presign(repo_id, bundle_id, size_bytes)
    return _pack_response(result, request)


@router.post(
    "/{owner}/{slug}/push/bundle/confirm",
    summary="Unpack a push bundle from MinIO and index into PostgreSQL",
    status_code=status.HTTP_200_OK,
)
@limiter.limit(OBJECT_LIMIT)
async def push_bundle_confirm(
    request: Request,
    owner: SlugParam,
    slug: SlugParam,
    claims: TokenClaims = Depends(require_valid_token),
    session: AsyncSession = Depends(get_session),
) -> Response:
    raw = await request.body()
    ct = request.headers.get("Content-Type", "")
    data = _decode_request_body(raw, ct)
    bundle_id = data.get("bundle_id", "")
    branch    = data.get("branch", "")
    head      = data.get("head", "")
    force     = bool(data.get("force", False))
    if not bundle_id or not branch or not head:
        raise HTTPException(status_code=422, detail="bundle_id, branch, head required")
    repo_id = await _resolve_repo_id(session, owner, slug)
    await _assert_writable(repo_id, claims, session)
    try:
        result = await wire_push_bundle_confirm(session, repo_id, bundle_id, branch, head, force)
    except ValueError as exc:
        raise HTTPException(status_code=422, detail=str(exc))
    return _pack_response(result, request)

Also add wire_push_bundle_presign and wire_push_bundle_confirm to the imports from musehub_wire.

FILE 3: muse/muse/cli/commands/push.py

Replace the body of _run_presign_path() (starting around line 421).

The data already in scope at that point:

• commits_oldest_first — list[dict] — all commits serialized
• snapshots_list — list[dict] — all snapshot manifests
• objects — list[ObjectPayload] — all file blobs with content bytes
• url — the remote base URL
• branch — target branch name
• local_head — the tip commit ID being pushed
• force — bool

New body:

async def _run_presign_path() -> "PushResult":
    import msgpack as _msgpack
    import hashlib as _hashlib
    import time as _tm

    bundle_presign_endpoint = f"{url.rstrip('/')}/push/bundle/presign"
    bundle_confirm_endpoint = f"{url.rstrip('/')}/push/bundle/confirm"

    # Build bundle dict from data already in memory.
    bundle_dict = {
        "commits":   commits_oldest_first,
        "snapshots": snapshots_list,
        "objects":   objects,
        "meta": {
            "mode": "incremental",
            "base_commits": _branch_have,
            "created_at": _time.strftime("%Y-%m-%dT%H:%M:%SZ", _time.gmtime()),
        },
    }
    bundle_bytes = _msgpack.packb(bundle_dict, use_bin_type=True)
    bundle_id = "sha256:" + _hashlib.sha256(bundle_bytes).hexdigest()
    print(
        f"  [bundle] packed {len(commits_oldest_first)} commits, "
        f"{len(snapshots_list)} snapshots, {len(objects)} objects → "
        f"{len(bundle_bytes)//1024}KiB bundle ({bundle_id[:16+7]}…)",
        file=sys.stderr,
    )

    async with _httpx.AsyncClient(
        http2=False, timeout=_PUSH_TIMEOUT,
        verify=_verify, limits=_limits,
    ) as _client:
        # Step 1: Get presigned URL for the whole bundle.
        presign_body = _msgpack.packb(
            {"bundle_id": bundle_id, "size_bytes": len(bundle_bytes)},
            use_bin_type=True,
        )
        _presign_headers = dict(transport._build_request(
            "POST", bundle_presign_endpoint, signing, presign_body,
            extra_headers={"Content-Type": "application/x-msgpack"},
        ).headers)
        _t_presign = _tm.perf_counter()
        presign_resp = await _client.post(
            bundle_presign_endpoint,
            content=presign_body,
            headers=_presign_headers,
        )
        if presign_resp.status_code >= 400:
            raise TransportError(
                f"push/bundle/presign: HTTP {presign_resp.status_code}: {presign_resp.text[:200]}",
                presign_resp.status_code,
            )
        presign_data = _msgpack.unpackb(presign_resp.content, raw=False)
        presigned_url = presign_data["presigned_url"]
        print(
            f"  [bundle] presign: {_tm.perf_counter()-_t_presign:.3f}s → got URL",
            file=sys.stderr,
        )

        # Step 2: Upload entire bundle in one PUT.
        _t_put = _tm.perf_counter()
        put_resp = await _client.put(
            presigned_url,
            content=bundle_bytes,
            headers={"Content-Type": "application/octet-stream"},
        )
        if put_resp.status_code not in (200, 204):
            raise TransportError(
                f"bundle PUT failed: HTTP {put_resp.status_code}",
                put_resp.status_code,
            )
        print(
            f"  [bundle] PUT {len(bundle_bytes)//1024}KiB: {_tm.perf_counter()-_t_put:.3f}s",
            file=sys.stderr,
        )

        # Step 3: Confirm — server unpacks, indexes, updates branch.
        confirm_body = _msgpack.packb(
            {"bundle_id": bundle_id, "branch": branch, "head": local_head, "force": force},
            use_bin_type=True,
        )
        _confirm_headers = dict(transport._build_request(
            "POST", bundle_confirm_endpoint, signing, confirm_body,
            extra_headers={"Content-Type": "application/x-msgpack"},
        ).headers)
        _t_confirm = _tm.perf_counter()
        confirm_resp = await _client.post(
            bundle_confirm_endpoint,
            content=confirm_body,
            headers=_confirm_headers,
        )
        if confirm_resp.status_code >= 400:
            raise TransportError(
                f"push/bundle/confirm: HTTP {confirm_resp.status_code}: {confirm_resp.text[:200]}",
                confirm_resp.status_code,
            )
        confirm_data = _msgpack.unpackb(confirm_resp.content, raw=False)
        print(
            f"  [bundle] confirm: {_tm.perf_counter()-_t_confirm:.3f}s "
            f"(commits={confirm_data.get('stored_commits')}, "
            f"objects={confirm_data.get('stored_objects')})",
            file=sys.stderr,
        )

        return PushResult(
            ok=True,
            message=f"pushed {confirm_data.get('stored_commits', 0)} commit(s) to {branch!r}",
            head=local_head,
            heads={branch: local_head},
            stored_commits=confirm_data.get("stored_commits", 0),
            stored_objects=confirm_data.get("stored_objects", 0),
            already_present_objects=0,
            code=0,
        )

result = await _run_presign_path()

Invariants the implementation must preserve

1. Content-addressing is never bypassed. bundle_id = sha256(bundle_bytes). Server verifies before unpacking.
2. Ed25519 signatures are verified server-side before any DB write.
3. Individual objects must still be written to MinIO by their sha256 key so /o/{object_id}, fetch, and presign_get all work.
4. on_conflict_do_nothing on all inserts — the operation is idempotent on retry.
5. Single session.commit() at the end — all writes are atomic.

What is NOT in scope for this ticket

• Fast-forward ancestry check on branch update (the existing wire_push_stream code has this; copy it in a follow-up)
• Quota checks (can be added inside wire_push_bundle_confirm in a follow-up)
• Removing push/presign + push/confirm legacy endpoints (keep them; delete in a separate ticket after bundle path is stable)
• Small push path (push/stream with inline O frames) — unchanged

Success criteria

time muse push local dev

On a fresh repo (MinIO and PG cleared), 1,024-commit / 5,139-object push:

Branch

Both repos are already branched at task/bundle-push.

• ~/ecosystem/muse — client changes go in muse/cli/commands/push.py
• ~/ecosystem/musehub — server changes go in musehub/services/musehub_wire.py and musehub/api/routes/wire.py

Do not touch the stream path. Do not remove legacy presign endpoints yet.