gabriel / musehub public
test_mpack_delta_e2e.py python
283 lines 9.5 KB
Raw
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595 fix: typing audit — 0 violations, 0 untyped defs across all… Sonnet 4.6 minor ⚠ breaking 49 days ago
1 """TDD — mpack push delta format end-to-end.
2
3 Proves the full path: client builds delta mpack → PUT to MinIO →
4 server reconstructs manifests from deltas → snapshots written correctly.
5
6 Dimensions match the real musehub repo: ~1031 commits, ~700 files per
7 snapshot, ~5 files changed per commit.
8
9 The one principle: snapshot_id = sha256(manifest). The delta chain is
10 the proof. No full manifest blobs on the wire or in PG.
11 """
12 from __future__ import annotations
13
14 import datetime
15 import hashlib
16 import pathlib
17 import time
18
19 import msgpack
20 import pytest
21 import pytest_asyncio
22 from httpx import AsyncClient, ASGITransport
23 from sqlalchemy.ext.asyncio import AsyncSession
24
25 from musehub.auth.request_signing import MSignContext, require_signed_request, optional_signed_request
26 from musehub.db.database import get_db
27 from musehub.main import app
28
29 from muse.core.object_store import write_object
30 from muse.core.mpack import build_mpack, build_wire_mpack
31 from muse.core.paths import muse_dir
32 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
33 from muse.core.commits import CommitRecord, write_commit
34 from muse.core.refs import write_branch_ref
35 from muse.core.snapshots import SnapshotRecord, write_snapshot
36 from muse.core.types import blob_id
37
38 pytestmark = pytest.mark.wire
39
40
41 # ---------------------------------------------------------------------------
42 # Dimensions — match real musehub repo
43 # ---------------------------------------------------------------------------
44
45 _N_FILES = 700 # files per snapshot
46 _N_COMMITS = 1_031
47 _FILES_CHANGED = 5 # files changed per commit
48 _BLOB_SIZE = 512
49 _GATE_S = 60.0
50
51
52 # ---------------------------------------------------------------------------
53 # Auth fixtures
54 # ---------------------------------------------------------------------------
55
56 _AUTH_CTX = MSignContext(
57 handle="gabriel",
58 identity_id="sha256:" + "0" * 64,
59 is_agent=False,
60 is_admin=True,
61 )
62
63
64 @pytest_asyncio.fixture()
65 async def client(db_session: AsyncSession) -> None:
66 async def _override_get_db() -> None:
67 yield db_session
68
69 app.dependency_overrides[get_db] = _override_get_db
70 app.dependency_overrides[require_signed_request] = lambda: _AUTH_CTX
71 app.dependency_overrides[optional_signed_request] = lambda: _AUTH_CTX
72
73 async with AsyncClient(
74 transport=ASGITransport(app=app),
75 base_url="https://localhost:1337",
76 ) as c:
77 yield c
78
79 app.dependency_overrides.clear()
80
81
82 @pytest_asyncio.fixture()
83 async def repo(client: AsyncClient) -> None:
84 resp = await client.post(
85 "/api/repos",
86 json={"owner": "gabriel", "name": "mpack-delta-e2e", "visibility": "public", "initialize": False},
87 )
88 assert resp.status_code in (200, 201), resp.text
89 data = resp.json()
90 yield data["slug"]
91 await client.delete(f"/api/repos/{data['repoId']}")
92
93
94 # ---------------------------------------------------------------------------
95 # Local repo builder
96 # ---------------------------------------------------------------------------
97
98 def _make_repo(tmp: pathlib.Path) -> pathlib.Path:
99 tmp.mkdir(parents=True, exist_ok=True)
100 dot = muse_dir(tmp)
101 dot.mkdir()
102 (dot / "repo.json").write_text('{"repo_id":"delta-e2e","owner":"gabriel"}')
103 for d in ("commits", "snapshots", "objects"):
104 (dot / d).mkdir()
105 (dot / "refs" / "heads").mkdir(parents=True)
106 (dot / "HEAD").write_text("ref: refs/heads/main\n")
107 (dot / "config.toml").write_text("")
108 return tmp
109
110
111 def _populate(repo: pathlib.Path) -> str:
112 blob_ids: list[str] = []
113 for i in range(_N_FILES):
114 data = f"base-{i:06d}".encode() + b"x" * _BLOB_SIZE
115 oid = blob_id(data)
116 write_object(repo, oid, data)
117 blob_ids.append(oid)
118
119 base_manifest: dict[str, str] = {
120 f"src/file_{i:04d}.py": blob_ids[i] for i in range(_N_FILES)
121 }
122
123 parent: str | None = None
124 tip = ""
125 ts = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
126
127 for i in range(_N_COMMITS):
128 manifest = dict(base_manifest)
129 for j in range(_FILES_CHANGED):
130 idx = (i * _FILES_CHANGED + j) % _N_FILES
131 variant = f"commit-{i:05d}-file-{j}".encode() + b"y" * _BLOB_SIZE
132 variant_oid = blob_id(variant)
133 write_object(repo, variant_oid, variant)
134 manifest[f"src/file_{idx:04d}.py"] = variant_oid
135
136 sid = compute_snapshot_id(manifest)
137 write_snapshot(repo, SnapshotRecord(snapshot_id=sid, manifest=manifest))
138
139 msg = f"commit-{i:05d}"
140 cid = compute_commit_id(
141 parent_ids=[parent] if parent else [],
142 snapshot_id=sid,
143 message=msg,
144 committed_at_iso=ts.isoformat(),
145 author="gabriel",
146 )
147 write_commit(repo, CommitRecord(
148 commit_id=cid,
149 branch="main",
150 snapshot_id=sid,
151 message=msg,
152 committed_at=ts,
153 parent_commit_id=parent,
154 parent2_commit_id=None,
155 author="gabriel",
156 metadata={},
157 structured_delta=None,
158 sem_ver_bump="none",
159 breaking_changes=[],
160 agent_id="", model_id="", toolchain_id="",
161 prompt_hash="", signature="", signer_key_id="",
162 ))
163 parent = cid
164 tip = cid
165 ts += datetime.timedelta(seconds=60)
166
167 write_branch_ref(repo, "main", tip)
168 return tip
169
170
171 # ---------------------------------------------------------------------------
172 # Tests
173 # ---------------------------------------------------------------------------
174
175 def test_mpack_is_delta_encoded(tmp_path: pathlib.Path) -> None:
176 """Client-side: mpack snapshots are deltas, not full manifests.
177
178 First snapshot has delta_upsert == full manifest (no parent).
179 All subsequent snapshots have delta_upsert << _N_FILES entries.
180 """
181 repo = _make_repo(tmp_path / "repo")
182 head = _populate(repo)
183 mpack = build_mpack(repo, [head], have=[])
184
185 snaps = mpack.get("snapshots") or []
186 assert len(snaps) == _N_COMMITS
187
188 # First snapshot: full manifest as delta_upsert (no parent)
189 assert len(snaps[0].get("delta_upsert", {})) == _N_FILES
190 assert snaps[0].get("parent_snapshot_id") is None
191
192 # All subsequent: only changed files
193 for snap in snaps[1:]:
194 n = len(snap.get("delta_upsert", {}))
195 # Each commit changes _FILES_CHANGED files; when those differ from the
196 # previous commit's changed files, delta_upsert includes both the new
197 # additions and the reversions — at most 2× _FILES_CHANGED.
198 assert n <= _FILES_CHANGED * 2, (
199 f"snapshot {snap['snapshot_id'][:16]} has {n} delta_upsert entries — "
200 f"expected ≤ {_FILES_CHANGED * 2}"
201 )
202 assert "manifest" not in snap, "full manifest must not appear in delta mpack"
203
204 # Wire size: delta snapshots must be < 5% of full-manifest equivalent
205 full_size = _N_COMMITS * _N_FILES * 80 # rough: 80 bytes per path+oid entry
206 delta_size = sum(
207 len(msgpack.packb(s, use_bin_type=True)) for s in snaps
208 )
209 ratio = delta_size / full_size
210 assert ratio < 0.05, f"delta snapshots are {ratio:.1%} of full — expected < 5%"
211
212
213 @pytest.mark.asyncio
214 async def test_mpack_push_delta_e2e(
215 client: AsyncClient, repo: str, tmp_path: pathlib.Path, db_session: AsyncSession
216 ) -> None:
217 """Full path: build delta mpack → store in backend → unpack-mpack → verify.
218
219 Gate: server pipeline < 60s for 1031 commits × 700 files × 5 changed per commit.
220 Proves manifest_blob is NOT stored (delta chain is the proof).
221 """
222 import musehub.storage.backends as _backends_mod
223 from muse.core.types import blob_id as _blob_id
224
225 local_repo = _make_repo(tmp_path / "repo")
226 head = _populate(local_repo)
227
228 mpack = build_mpack(local_repo, [head], have=[])
229 wire_bytes = build_wire_mpack(mpack)
230 mpack_key = _blob_id(wire_bytes)
231
232 backend = _backends_mod.get_backend()
233 await backend.put_mpack(mpack_key, wire_bytes)
234
235 n_commits = len(mpack.get("commits") or [])
236 n_blobs = len(mpack.get("blobs") or [])
237
238 t_start = time.perf_counter()
239
240 unpack_resp = await client.post(
241 f"/gabriel/{repo}/push/unpack-mpack",
242 content=msgpack.packb(
243 {
244 "mpack_key": mpack_key,
245 "branch": "main",
246 "head": head,
247 "commits_count": n_commits,
248 "blobs_count": n_blobs,
249 },
250 use_bin_type=True,
251 ),
252 headers={"Content-Type": "application/x-msgpack"},
253 )
254 assert unpack_resp.status_code == 200, unpack_resp.text
255 result = unpack_resp.json()
256
257 t_unpack = time.perf_counter()
258
259 refs_resp = await client.get(f"/gabriel/{repo}/refs")
260 assert refs_resp.status_code == 200
261 assert refs_resp.json().get("branch_heads", {}).get("main") == head
262
263 t_done = time.perf_counter()
264
265 server_ms = (t_done - t_start) * 1000
266
267 assert result.get("commits_written") == _N_COMMITS, result
268 assert result.get("snapshots_written") == _N_COMMITS, result
269
270 gate_ms = _GATE_S * 1000
271 assert server_ms < gate_ms, (
272 f"Gate FAIL: {server_ms:.0f}ms > {gate_ms:.0f}ms\n"
273 f" unpack={(t_unpack-t_start)*1000:.0f}ms "
274 f"refs={(t_done-t_unpack)*1000:.0f}ms"
275 )
276
277 print(
278 f"\n {_N_COMMITS} commits × {_N_FILES} files × {_FILES_CHANGED} changed/commit\n"
279 f" mpack wire: {len(wire_bytes)//1024} KiB\n"
280 f" server total: {server_ms:.0f}ms (gate {gate_ms:.0f}ms)\n"
281 f" commits: {result.get('commits_written')}\n"
282 f" snapshots: {result.get('snapshots_written')}"
283 )
File History 1 commit
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595 fix: typing audit — 0 violations, 0 untyped defs across all… Sonnet 4.6 minor ⚠ 49 days ago