gabriel / musehub public
test_mpack_push_async.py python
353 lines 12.2 KB
Raw
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595 fix: typing audit — 0 violations, 0 untyped defs across all… Sonnet 4.6 minor ⚠ breaking 20 days ago
1 """TDD — mpack push is async: fast sync path + background indexing.
2
3 The one principle: Content-addressing is a proof, not a label.
4
5 The mpack is in MinIO. sha256(wire_bytes) == mpack_key is the proof.
6 Everything that follows — commit rows, snapshot rows — is derived indexing.
7 Derived indexing MUST NOT block the synchronous push response.
8
9 Synchronous path (must complete < 1s for real-world repo):
10 1. GET mpack from MinIO
11 2. Verify sha256(wire_bytes) == mpack_key
12 3. Advance branch pointer
13 4. Enqueue mpack.index job
14 5. Return 200
15
16 Background path (mpack.index job):
17 1. GET mpack from MinIO
18 2. Reconstruct snapshot deltas → INSERT musehub_snapshots
19 3. INSERT commits → musehub_commits
20 4. Return counts
21
22 Dimensions match the real musehub repo: ~1031 commits, ~700 files per
23 snapshot, ~5 files changed per commit.
24 """
25 from __future__ import annotations
26
27 import datetime
28 import hashlib
29 import pathlib
30 import time
31
32 import httpx
33 import msgpack
34 import pytest
35 import pytest_asyncio
36
37 pytestmark = pytest.mark.skip(reason="muse wire protocol in flux")
38 from httpx import AsyncClient, ASGITransport
39 from sqlalchemy import select
40 from sqlalchemy.ext.asyncio import AsyncSession
41
42 from musehub.auth.request_signing import MSignContext, require_signed_request, optional_signed_request
43 from musehub.db.musehub_repo_models import MusehubCommit, MusehubCommitRef, MusehubRepo, MusehubSnapshot, MusehubSnapshotRef
44 from musehub.db.database import get_db
45 from musehub.main import app
46
47 from muse.core.object_store import write_object
48 from muse.core.mpack import build_mpack
49 from muse.core.paths import muse_dir
50 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
51 from muse.core.commits import CommitRecord, write_commit
52 from muse.core.refs import write_branch_ref
53 from muse.core.snapshots import SnapshotRecord, write_snapshot
54 from muse.core.types import blob_id
55
56
57 # ---------------------------------------------------------------------------
58 # Dimensions — match real musehub repo
59 # ---------------------------------------------------------------------------
60
61 _N_FILES = 700
62 _N_COMMITS = 1_031
63 _FILES_CHANGED = 5
64 _BLOB_SIZE = 512
65
66 # unpack-mpack does full indexing synchronously; gate is wall-clock for 1031 commits
67 _SYNC_GATE_S = 10.0
68
69
70 # ---------------------------------------------------------------------------
71 # Auth + fixtures
72 # ---------------------------------------------------------------------------
73
74 _AUTH_CTX = MSignContext(
75 handle="gabriel",
76 identity_id="sha256:" + "0" * 64,
77 is_agent=False,
78 is_admin=True,
79 )
80
81
82 @pytest_asyncio.fixture()
83 async def client(db_session: AsyncSession) -> None:
84 async def _override_get_db() -> None:
85 yield db_session
86
87 app.dependency_overrides[get_db] = _override_get_db
88 app.dependency_overrides[require_signed_request] = lambda: _AUTH_CTX
89 app.dependency_overrides[optional_signed_request] = lambda: _AUTH_CTX
90
91 async with AsyncClient(
92 transport=ASGITransport(app=app),
93 base_url="https://localhost:1337",
94 ) as c:
95 yield c
96
97 app.dependency_overrides.clear()
98
99
100 @pytest_asyncio.fixture()
101 async def repo(client: AsyncClient) -> None:
102 resp = await client.post(
103 "/api/repos",
104 json={"owner": "gabriel", "name": "mpack-async-e2e", "visibility": "public", "initialize": False},
105 )
106 assert resp.status_code in (200, 201), resp.text
107 data = resp.json()
108 yield data["slug"]
109 await client.delete(f"/api/repos/{data['repoId']}")
110
111
112 # ---------------------------------------------------------------------------
113 # Local repo builder (shared with test_mpack_delta_e2e)
114 # ---------------------------------------------------------------------------
115
116 def _make_repo(tmp: pathlib.Path) -> pathlib.Path:
117 tmp.mkdir(parents=True, exist_ok=True)
118 dot = muse_dir(tmp)
119 dot.mkdir()
120 (dot / "repo.json").write_text('{"repo_id":"async-e2e","owner":"gabriel"}')
121 for d in ("commits", "snapshots", "objects"):
122 (dot / d).mkdir()
123 (dot / "refs" / "heads").mkdir(parents=True)
124 (dot / "HEAD").write_text("ref: refs/heads/main\n")
125 (dot / "config.toml").write_text("")
126 return tmp
127
128
129 def _populate(repo: pathlib.Path) -> str:
130 blob_ids: list[str] = []
131 for i in range(_N_FILES):
132 data = f"base-{i:06d}".encode() + b"x" * _BLOB_SIZE
133 oid = blob_id(data)
134 write_object(repo, oid, data)
135 blob_ids.append(oid)
136
137 base_manifest: dict[str, str] = {
138 f"src/file_{i:04d}.py": blob_ids[i] for i in range(_N_FILES)
139 }
140
141 parent: str | None = None
142 tip = ""
143 ts = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
144
145 for i in range(_N_COMMITS):
146 manifest = dict(base_manifest)
147 for j in range(_FILES_CHANGED):
148 idx = (i * _FILES_CHANGED + j) % _N_FILES
149 variant = f"commit-{i:05d}-file-{j}".encode() + b"y" * _BLOB_SIZE
150 variant_oid = blob_id(variant)
151 write_object(repo, variant_oid, variant)
152 manifest[f"src/file_{idx:04d}.py"] = variant_oid
153
154 sid = compute_snapshot_id(manifest)
155 write_snapshot(repo, SnapshotRecord(snapshot_id=sid, manifest=manifest))
156
157 msg = f"commit-{i:05d}"
158 cid = compute_commit_id(
159 parent_ids=[parent] if parent else [],
160 snapshot_id=sid,
161 message=msg,
162 committed_at_iso=ts.isoformat(),
163 author="gabriel",
164 )
165 write_commit(repo, CommitRecord(
166 commit_id=cid,
167 branch="main",
168 snapshot_id=sid,
169 message=msg,
170 committed_at=ts,
171 parent_commit_id=parent,
172 parent2_commit_id=None,
173 author="gabriel",
174 metadata={},
175 structured_delta=None,
176 sem_ver_bump="none",
177 breaking_changes=[],
178 agent_id="", model_id="", toolchain_id="",
179 prompt_hash="", signature="", signer_key_id="",
180 ))
181 parent = cid
182 tip = cid
183 ts += datetime.timedelta(seconds=60)
184
185 write_branch_ref(repo, "main", tip)
186 return tip
187
188
189 # ---------------------------------------------------------------------------
190 # Helpers
191 # ---------------------------------------------------------------------------
192
193 async def _upload_mpack(client: AsyncClient, repo_slug: str, wire_bytes: bytes) -> str:
194 mpack_key = "sha256:" + hashlib.sha256(wire_bytes).hexdigest()
195
196 presign_resp = await client.post(
197 f"/gabriel/{repo_slug}/push/mpack-presign",
198 content=msgpack.packb(
199 {"mpack_key": mpack_key, "size_bytes": len(wire_bytes)},
200 use_bin_type=True,
201 ),
202 headers={"Content-Type": "application/x-msgpack"},
203 )
204 assert presign_resp.status_code == 200, presign_resp.text
205 upload_url = presign_resp.json().get("upload_url") or presign_resp.json().get("uploadUrl")
206
207 async with httpx.AsyncClient() as raw:
208 put_resp = await raw.put(upload_url, content=wire_bytes)
209 assert put_resp.status_code in (200, 204)
210
211 return mpack_key
212
213
214 # ---------------------------------------------------------------------------
215 # Tests
216 # ---------------------------------------------------------------------------
217
218 @pytest.mark.asyncio
219 async def test_unpack_mpack_returns_fast(
220 client: AsyncClient, repo: str, tmp_path: pathlib.Path
221 ) -> None:
222 """unpack-mpack must complete within the gate for the full repo.
223
224 The synchronous path does all indexing inline (commits, snapshots, objects,
225 branch advance). This gate proves the full pipeline stays within an
226 acceptable wall-clock budget for the real-world repo dimensions.
227 """
228 local_repo = _make_repo(tmp_path / "repo")
229 head = _populate(local_repo)
230
231 mpack = build_mpack(local_repo, [head], have=[])
232 wire_bytes = msgpack.packb(mpack, use_bin_type=True)
233
234 mpack_key = await _upload_mpack(client, repo, wire_bytes)
235
236 t0 = time.perf_counter()
237 unpack_resp = await client.post(
238 f"/gabriel/{repo}/push/unpack-mpack",
239 content=msgpack.packb(
240 {"mpack_key": mpack_key, "branch": "main", "head": head},
241 use_bin_type=True,
242 ),
243 headers={"Content-Type": "application/x-msgpack"},
244 )
245 elapsed = time.perf_counter() - t0
246
247 assert unpack_resp.status_code == 200, unpack_resp.text
248 result = unpack_resp.json()
249
250 # Branch pointer must be set
251 refs_resp = await client.get(f"/gabriel/{repo}/refs")
252 assert refs_resp.status_code == 200
253 assert refs_resp.json().get("branch_heads", {}).get("main") == head
254
255 # Commits and snapshots must be written inline
256 assert result.get("commits_written") == _N_COMMITS, result
257 assert result.get("snapshots_written") == _N_COMMITS, result
258
259 assert elapsed < _SYNC_GATE_S, (
260 f"unpack-mpack took {elapsed:.2f}s — gate is {_SYNC_GATE_S}s\n"
261 f" {_N_COMMITS} commits × {_N_FILES} files × {_FILES_CHANGED} changed/commit"
262 )
263
264 print(
265 f"\n {_N_COMMITS} commits × {_N_FILES} files × {_FILES_CHANGED} changed/commit\n"
266 f" sync elapsed: {elapsed*1000:.0f}ms (gate {_SYNC_GATE_S*1000:.0f}ms)\n"
267 f" commits: {result.get('commits_written')}\n"
268 f" snapshots: {result.get('snapshots_written')}"
269 )
270
271
272 @pytest.mark.asyncio
273 async def test_mpack_index_job_populates_commits_and_snapshots(
274 client: AsyncClient, repo: str, tmp_path: pathlib.Path, db_session: AsyncSession
275 ) -> None:
276 """All commits and snapshots must be in PG after unpack-mpack returns.
277
278 Indexing is synchronous — the route writes commits, snapshots, and objects
279 inline before returning 200. This test verifies the DB state is correct.
280 """
281 local_repo = _make_repo(tmp_path / "repo")
282 head = _populate(local_repo)
283
284 mpack = build_mpack(local_repo, [head], have=[])
285 wire_bytes = msgpack.packb(mpack, use_bin_type=True)
286
287 mpack_key = await _upload_mpack(client, repo, wire_bytes)
288
289 unpack_resp = await client.post(
290 f"/gabriel/{repo}/push/unpack-mpack",
291 content=msgpack.packb(
292 {"mpack_key": mpack_key, "branch": "main", "head": head},
293 use_bin_type=True,
294 ),
295 headers={"Content-Type": "application/x-msgpack"},
296 )
297 assert unpack_resp.status_code == 200, unpack_resp.text
298 result = unpack_resp.json()
299
300 assert result.get("commits_written") == _N_COMMITS, result
301 assert result.get("snapshots_written") == _N_COMMITS, result
302
303 # Verify rows are actually in PG
304 repo_row = (await db_session.execute(
305 select(MusehubRepo).where(MusehubRepo.slug == repo)
306 )).scalar_one()
307
308 commit_count = (await db_session.execute(
309 select(MusehubCommit)
310 .join(MusehubCommitRef, MusehubCommitRef.commit_id == MusehubCommit.commit_id)
311 .where(MusehubCommitRef.repo_id == repo_row.repo_id)
312 )).scalars().all()
313 assert len(commit_count) >= _N_COMMITS
314
315 snap_count = (await db_session.execute(
316 select(MusehubSnapshot)
317 .join(MusehubSnapshotRef, MusehubSnapshotRef.snapshot_id == MusehubSnapshot.snapshot_id)
318 .where(MusehubSnapshotRef.repo_id == repo_row.repo_id)
319 )).scalars().all()
320 assert len(snap_count) >= _N_COMMITS
321
322
323 @pytest.mark.asyncio
324 async def test_branch_head_correct_before_indexing(
325 client: AsyncClient, repo: str, tmp_path: pathlib.Path
326 ) -> None:
327 """Branch pointer is correct immediately after unpack-mpack, before worker runs.
328
329 This is the fundamental correctness guarantee: push is 'done' once the
330 branch pointer is updated. The indexing is eventual but the ref is immediate.
331 """
332 local_repo = _make_repo(tmp_path / "repo")
333 head = _populate(local_repo)
334
335 mpack = build_mpack(local_repo, [head], have=[])
336 wire_bytes = msgpack.packb(mpack, use_bin_type=True)
337
338 mpack_key = await _upload_mpack(client, repo, wire_bytes)
339
340 unpack_resp = await client.post(
341 f"/gabriel/{repo}/push/unpack-mpack",
342 content=msgpack.packb(
343 {"mpack_key": mpack_key, "branch": "main", "head": head},
344 use_bin_type=True,
345 ),
346 headers={"Content-Type": "application/x-msgpack"},
347 )
348 assert unpack_resp.status_code == 200, unpack_resp.text
349
350 # Ref must be set — worker has NOT run yet
351 refs_resp = await client.get(f"/gabriel/{repo}/refs")
352 assert refs_resp.status_code == 200
353 assert refs_resp.json()["branch_heads"]["main"] == head
File History 1 commit
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595 fix: typing audit — 0 violations, 0 untyped defs across all… Sonnet 4.6 minor ⚠ 20 days ago