gabriel / musehub public
feat BREAKING wire task/streaming-object-fetch #1 / 1
gabriel · 163 days ago · Apr 15, 2026 · Diff

feat(wire): stream fetch/objects via msgpack-stream for TTFO improvement

Replace buffered _pack_objects_response with _stream_objects async generator using asyncio.as_completed — objects are written to the response as each R2 GET completes instead of waiting for all fetches.

Changes: - New _MSGPACK_STREAM_CT = 'application/x-msgpack-stream' content type - New _stream_objects() async generator (asyncio.as_completed, 32-worker semaphore) - fetch_objects route now returns StreamingResponse(_stream_objects(...)) - 23 new tests across 8 tiers in test_wire_object_streaming.py

sha256:93ca3b437a654dd019ba1fd795bf30877d35628eb95ace1823db9bc76a92132f sha
sha256:2af4343389565ac145e376403ce4d851f75be40948463fb7f2ddd68e458cf2f7 parent
+61 ~1 symbols
sha256:0d42b37f3d22b912ad228f3bb8ae012a4c2acf028511e8e1cef66f98b4d0f55c snapshot
+61
symbols added
~1
symbol modified
0
dead code introduced
Semantic Changes 62 symbols
~ tests/test_wire_object_streaming.py .py 59 symbols added
+ TestDocstrings class class TestDocstrings L610–623
+ test_fetch_objects_route_has_docstring method method test_fetch_objects_route_has_docstring L616–618
+ test_msgpack_stream_ct_constant_documented_in_module method method test_msgpack_stream_ct_constant_documented_in_module L620–623
+ test_stream_objects_has_docstring method method test_stream_objects_has_docstring L613–614
+ TestE2E class class TestE2E L424–483
+ test_empty_object_ids_returns_200_empty_body method async_method test_empty_object_ids_returns_200_empty_body L451–465
+ test_private_repo_unauthenticated_returns_403 method async_method test_private_repo_unauthenticated_returns_403 L467–483
+ test_push_then_stream_fetch_roundtrip method async_method test_push_then_stream_fetch_roundtrip L427–449
+ TestIntegration class class TestIntegration L298–345
+ test_multiple_objects_all_received method async_method test_multiple_objects_all_received L301–324
+ test_partial_oids_returns_only_known method async_method test_partial_oids_returns_only_known L326–345
+ TestPerformance class class TestPerformance L489–536
+ test_20_objects_under_200ms method async_method test_20_objects_under_200ms L513–536
+ test_single_object_under_50ms method async_method test_single_object_under_50ms L492–511
+ TestSchema class class TestSchema L213–292
+ test_each_object_has_required_keys method async_method test_each_object_has_required_keys L235–255
+ test_fetch_objects_returns_msgpack_stream_content_type method async_method test_fetch_objects_returns_msgpack_stream_content_type L216–233
+ test_object_content_matches_pushed_bytes method async_method test_object_content_matches_pushed_bytes L257–274
+ test_unknown_oid_produces_empty_stream method async_method test_unknown_oid_produces_empty_stream L276–292
+ TestSecurity class class TestSecurity L542–604
+ test_corrupt_object_omitted_from_stream method async_method test_corrupt_object_omitted_from_stream L545–564
+ test_empty_object_ids_accepted method async_method test_empty_object_ids_accepted L566–580
+ test_unauthenticated_public_repo_fetch_succeeds method async_method test_unauthenticated_public_repo_fetch_succeeds L582–604
+ TestStress class class TestStress L351–418
+ test_100_objects_all_received method async_method test_100_objects_all_received L354–372
+ test_stream_arrives_before_slowest_r2_fetch method async_method test_stream_arrives_before_slowest_r2_fetch L374–418
+ TestUnit class class TestUnit L174–207
+ test_concatenated_msgpack_parseable_by_unpacker method method test_concatenated_msgpack_parseable_by_unpacker L189–196
+ test_empty_obj_map_yields_nothing method method test_empty_obj_map_yields_nothing L181–187
+ test_msgpack_unpacker_partial_feed method method test_msgpack_unpacker_partial_feed L198–207
+ test_stream_content_type_constant_value method method test_stream_content_type_constant_value L177–179
+ _OWNER variable variable _OWNER L42–42
+ _mp function function _mp L55–56
+ _parse_stream function function _parse_stream L59–63
+ _push_object function async_function _push_object L71–116
+ _push_objects_batch function async_function _push_objects_batch L119–168
+ _sha256_oid function function _sha256_oid L51–52
+ _stream_headers function function _stream_headers L66–68
+ _utc_now function function _utc_now L47–48
+ AsyncClient import import AsyncClient L31–31
+ AsyncMock import import AsyncMock L27–27
+ AsyncSession import import AsyncSession L32–32
+ JSONObject import import JSONObject L37–37
+ _MSGPACK_STREAM_CT import import _MSGPACK_STREAM_CT L36–36
+ _stream_objects import import _stream_objects L36–36
+ annotations import import annotations L19–19
+ asyncio import import asyncio L21–21
+ datetime import import datetime L26–26
+ db import import db L34–34
+ factory_create_repo import import factory_create_repo L35–35
+ hashlib import import hashlib L22–22
+ inspect import import inspect L23–23
+ msgpack import import msgpack L29–29
+ patch import import patch L27–27
+ pytest import import pytest L30–30
+ time import import time L24–24
+ timezone import import timezone L26–26
+ uuid import import uuid L25–25
+ pytestmark variable variable pytestmark L39–39
~ musehub/api/routes/wire.py .py 2 symbols added, 1 symbol modified
+ _MSGPACK_STREAM_CT variable variable _MSGPACK_STREAM_CT L161–161
+ _stream_objects function async_function _stream_objects L164–249
~ fetch_objects
← Older Oldest on task/streaming-object-fetch
All commits
Newer → Latest on task/streaming-object-fetch

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:93ca3b437a654dd019ba1fd795bf30877d35628eb95ace1823db9bc76a92132f --body "your comment"