gabriel / musehub public
perf BREAKING task/perf-push-pipeline #1 / 1
gabriel · 172 days ago · Apr 9, 2026 · Diff

perf: eliminate O(N×M) snapshot row explosion and parallelise object uploads

Wire push pipeline now runs at GitHub scale:

1. manifest_blob (migration 0031): snapshot manifests stored as a single msgpack BYTEA blob per snapshot instead of N×M rows in snapshot_entries. 906 commits × 4000 files = 3.6M rows → 906 rows.

2. wire_push_objects bulk INSERT: replaced N session.add() + N/50 flush() with pg_insert(...).on_conflict_do_nothing() — one round-trip for any batch size.

3. SHA-256 verification off event loop: asyncio.to_thread() so concurrent chunk uploads are not serialised by CPU-bound hashing.

4. Concurrent storage: asyncio.gather() writes all new object bytes in parallel instead of sequentially.

5. Symbol indexer OOM cap (_MAX_INDEX_COMMITS_PER_PUSH=50): prevents server crash on first push of a large repo while job queue is built.

Tests: 5 new perf regression tests enforce concrete latency budgets: - bulk_upsert 100×500 snapshots < 3s - bulk_upsert 1000×100 snapshots < 5s - wire_push_objects 500 new objects < 1s - wire_push_objects 500 existing objects < 200ms (1 SELECT IN, zero writes) - 200-commit push does not OOM server

sha256:e649c062850aae26b3d40684a22d5bda0443feca8f348bb6f37ddf6046e965e7 sha
sha256:f463fbeb3a6726a1edfa7b8eac72c76d1897052b0d5a40cdacf4e304de80d92d parent
+42 ~10 −2 symbols
sha256:7bdb93d56070d556afb24d4e3dfa0684cbba52ae00af88c382ec8bf4cab80910 snapshot
+42
symbols added
~10
symbols modified
−2
symbols removed
0
dead code introduced
Semantic Changes 54 symbols
~ alembic/versions/0031_snapshot_manifest_blob.py .py 9 symbols added
+ branch_labels variable variable branch_labels L22–22
+ depends_on variable variable depends_on L23–23
+ down_revision variable variable down_revision L21–21
+ downgrade function function downgrade L38–39
+ annotations import import annotations L15–15
+ op import import op L18–18
+ sa import import sa L17–17
+ revision variable variable revision L20–20
+ upgrade function function upgrade L26–35
~ tests/test_perf_push_pipeline.py .py 28 symbols added
+ _make_snapshots function function _make_snapshots L54–68
+ _make_wire_objects function function _make_wire_objects L71–78
+ _mp function function _mp L81–82
+ _obj_id function function _obj_id L50–51
+ _snap_id function function _snap_id L46–47
+ _utc_now function function _utc_now L42–43
+ AsyncClient import import AsyncClient L30–30
+ AsyncSession import import AsyncSession L31–31
+ WireObject import import WireObject L34–34
+ WireObjectsRequest import import WireObjectsRequest L34–34
+ annotations import import annotations L20–20
+ asyncio import import asyncio L22–22
+ bulk_upsert_snapshot_entries import import bulk_upsert_snapshot_entries L35–35
+ datetime import import datetime L26–26
+ db import import db L33–33
+ factory_create_repo import import factory_create_repo L37–37
+ hashlib import import hashlib L23–23
+ msgpack import import msgpack L28–28
+ pytest import import pytest L29–29
+ time import import time L24–24
+ timezone import import timezone L26–26
+ uuid import import uuid L25–25
+ wire_push_objects import import wire_push_objects L36–36
+ test_push_pack_1000_commits_100_files_under_5s function async_function test_push_pack_1000_commits_100_files_under_5s L111–129
+ test_push_pack_100_commits_500_files_under_3s function async_function test_push_pack_100_commits_500_files_under_3s L88–107
+ test_symbol_indexer_large_push_does_not_crash function async_function test_symbol_indexer_large_push_does_not_crash L202–274
+ test_wire_push_objects_500_existing_objects_under_200ms function async_function test_wire_push_objects_500_existing_objects_under_200ms L162–196
+ test_wire_push_objects_500_new_objects_under_1s function async_function test_wire_push_objects_500_new_objects_under_1s L135–158
~ musehub/api/routes/wire.py .py 1 symbol added, 1 symbol modified
+ _MAX_INDEX_COMMITS_PER_PUSH variable variable _MAX_INDEX_COMMITS_PER_PUSH L794–794
~ _build_symbol_index_async
~ musehub/db/musehub_models.py .py 1 symbol modified
~ MusehubSnapshot
~ musehub/services/musehub_snapshot.py .py 2 symbols added, 2 symbols removed, 6 symbols modified
− delete import import delete L27–27
− insert import import insert L27–27
+ _decode_manifest_blob function function _decode_manifest_blob L237–242
+ msgpack import import msgpack L32–32
~ bulk_upsert_snapshot_entries
~ get_snapshot_manifest
~ get_snapshot_manifests_batch
~ func
~ select
~ upsert_snapshot_entries
~ musehub/services/musehub_wire.py .py 2 symbols added, 1 symbol modified
+ asyncio import import asyncio L22–22
+ pg_insert import import pg_insert L29–29
~ wire_push_objects
~ tests/test_wire_protocol.py .py 1 symbol modified
~ test_bulk_upsert_snapshot_entries_skips_existing
← Older Oldest on task/perf-push-pipeline
All commits
Newer → Latest on task/perf-push-pipeline

0 comments

No comments yet. Be the first to start the discussion.

To add a comment, use the Muse CLI: muse hub commit comment sha256:e649c062850aae26b3d40684a22d5bda0443feca8f348bb6f37ddf6046e965e7 --body "your comment"