gabriel / musehub public
test_push_timeout_fix.py python
274 lines 10.0 KB
Raw
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 122 days ago
1 """TDD — push must stream response bytes so Cloudflare doesn't 524-timeout.
2
3 Root cause of the large-repo push failure (2026-04-28):
4 _WireResponse buffered ALL wire_push_stream frames before sending a single
5 HTTP response byte. CF's 120-second origin timeout fired during the silent
6 commit-processing phase (827 commits × DB inserts ≈ 8+ minutes).
7
8 Three tests codify the correct behaviour:
9
10 T1 Static: _WireResponse must send http.response.start BEFORE it has
11 consumed all frames from wire_push_stream. Sending headers early
12 resets CF's 120-second "first byte" timer.
13
14 T2 Integration: Push response body must be a sequence of msgpack objects:
15 zero or more PROGRESS frames followed by one RESULT frame.
16 (The old response was a single msgpack dict with no "t" field.)
17
18 T3 Static: wire_push_stream must yield PROGRESS frames INSIDE the commit
19 insertion loop — not just the single frame emitted before the entire
20 loop. Without intra-loop heartbeats, CF still timeouts even after T1.
21 """
22 from __future__ import annotations
23
24 import inspect
25 import textwrap
26
27 import msgpack
28 import pytest
29 from httpx import AsyncClient
30 from sqlalchemy.ext.asyncio import AsyncSession
31
32 from muse.core.types import blob_id
33 from muse.core.mpack import WIRE_CONTENT_TYPE, MuseWireFrameWriter
34 from musehub.models.wire import (
35 SFRAME_COMMIT_PACK,
36 SFRAME_END,
37 SFRAME_ERROR,
38 SFRAME_HEADER,
39 SFRAME_OBJECT,
40 SFRAME_PROGRESS,
41 SFRAME_RESULT,
42 )
43 from musehub.types.json_types import JSONObject, JSONValue, StrDict
44 from tests.factories import create_repo
45
46 _fw = MuseWireFrameWriter()
47
48
49 # ---------------------------------------------------------------------------
50 # Helpers
51 # ---------------------------------------------------------------------------
52
53 def _pack(obj: JSONValue) -> bytes:
54 return msgpack.packb(obj, use_bin_type=True)
55
56
57 def _wrap(ft: str, data: JSONValue) -> bytes:
58 return _fw.wrap(frame_type=ft, payload=_pack(data))
59
60
61 def _oid(data: bytes) -> str:
62 return blob_id(data)
63
64
65 def _header_frame(n_objects: int = 0, n_commits: int = 1) -> bytes:
66 return _wrap(SFRAME_HEADER, {
67 "t": SFRAME_HEADER,
68 "branch": "main",
69 "force": False,
70 "have": [],
71 "head": _oid(b"head"),
72 "n_objects": n_objects,
73 "n_commits": n_commits,
74 })
75
76
77 def _object_frame(raw: bytes, path: str = "file.py") -> tuple[str, bytes]:
78 oid = _oid(raw)
79 return oid, _wrap(SFRAME_OBJECT, {
80 "t": SFRAME_OBJECT,
81 "id": oid,
82 "path": path,
83 "enc": "raw",
84 "content": raw,
85 })
86
87
88 def _commit_pack_frame(commits: list[JSONObject], snapshots: list[JSONObject]) -> bytes:
89 return _wrap(SFRAME_COMMIT_PACK, {
90 "t": SFRAME_COMMIT_PACK,
91 "commits": commits,
92 "snapshots": snapshots,
93 })
94
95
96 def _end_frame(n_objects: int = 0, n_commits: int = 1) -> bytes:
97 return _wrap(SFRAME_END, {"t": SFRAME_END, "n_objects": n_objects, "n_commits": n_commits})
98
99
100 def _make_commit(
101 snapshot_id: str,
102 parent_id: str | None = None,
103 suffix: str = "",
104 ) -> JSONObject:
105 return {
106 "commit_id": _oid(f"commit-{snapshot_id}{suffix}".encode()),
107 "parent_ids": [parent_id] if parent_id else [],
108 "snapshot_id": snapshot_id,
109 "branch": "main",
110 "message": "timeout tdd commit",
111 "author": "test-user-wire",
112 "committed_at": "2026-04-28T00:00:00+00:00",
113 "signature": "",
114 "signer_key_id": "",
115 "agent_id": "claude-code",
116 "model_id": "claude-sonnet-4-6",
117 "metadata": {},
118 }
119
120
121 def _make_snapshot(snap_id: str, manifest: StrDict | None = None) -> JSONObject:
122 return {"snapshot_id": snap_id, "manifest": manifest or {}}
123
124
125 def _parse_response_frames(content: bytes) -> list[dict]:
126 """Parse response body as a sequence of msgpack objects."""
127 unpacker = msgpack.Unpacker(raw=False)
128 unpacker.feed(content)
129 return list(unpacker)
130
131
132 # ---------------------------------------------------------------------------
133 # T1 — _WireResponse must send http.response.start BEFORE consuming all frames
134 # ---------------------------------------------------------------------------
135
136 def test_t1_wire_response_sends_headers_before_consuming_full_stream() -> None:
137 """_WireResponse.__call__ must call send(http.response.start) before the
138 async-for loop over wire_push_stream completes.
139
140 The current broken pattern:
141 async for frame in wire_push_stream(...): ...
142 await send({"type": "http.response.start", ...}) # ← happens last
143
144 The correct pattern:
145 await send({"type": "http.response.start", ...}) # ← happens first
146 async for frame in wire_push_stream(...):
147 await send({"type": "http.response.body", "body": frame, "more_body": True})
148 await send({"type": "http.response.body", "body": b"", "more_body": False})
149
150 This test checks the source of push_stream for the correct ordering.
151 """
152 from musehub.api.routes import wire
153
154 source = inspect.getsource(wire.push_stream)
155
156 # The response.start send must happen before any frame iteration.
157 # In the correct implementation, http.response.start appears before the
158 # async for loop; in the broken implementation it appears after.
159 start_pos = source.find("http.response.start")
160 loop_pos = source.find("async for frame_bytes in wire_push_stream")
161
162 assert start_pos != -1, (
163 "push_stream source must contain 'http.response.start'"
164 )
165 assert loop_pos != -1, (
166 "push_stream source must contain 'async for frame_bytes in wire_push_stream'"
167 )
168 assert start_pos < loop_pos, (
169 "http.response.start must be sent BEFORE the async for loop over wire_push_stream. "
170 "Currently the send happens after all frames are consumed — CF's 120-second timer "
171 "fires during commit processing because no bytes reach CF until then. "
172 "Fix: send http.response.start early, then stream P frames as body chunks."
173 )
174
175
176 # ---------------------------------------------------------------------------
177 # T2 — push response body is a stream of msgpack objects (P* then R)
178 # ---------------------------------------------------------------------------
179
180 @pytest.mark.asyncio
181 async def test_t2_push_response_is_frame_stream(
182 client: AsyncClient,
183 db_session: AsyncSession,
184 wire_headers: StrDict,
185 ) -> None:
186 """Push response body must be a sequence of msgpack objects ending in a
187 RESULT frame, not a single opaque msgpack dict.
188
189 The old response: one dict { "ok": True, "message": "...", ... } (no "t" key)
190 The new response: N dicts with "t" == "P", then one dict with "t" == "R"
191
192 This is what the protocol spec says ("The server responds with a matching
193 frame stream: PROGRESS frames, ERROR frame or RESULT frame"). The
194 streaming format lets CF see progress bytes and avoids the 524 timeout.
195 """
196 repo = await create_repo(db_session, owner="test-user-wire", name="t2-frame-stream")
197
198 snap_id = _oid(b"snap-t2")
199 commit = _make_commit(snap_id)
200 snap = _make_snapshot(snap_id)
201
202 body = (
203 _header_frame(n_objects=0, n_commits=1)
204 + _commit_pack_frame([commit], [snap])
205 + _end_frame(n_objects=0, n_commits=1)
206 )
207
208 resp = await client.post(
209 f"/{repo.owner}/{repo.slug}/push/stream",
210 content=body,
211 headers={**wire_headers, "Content-Type": WIRE_CONTENT_TYPE},
212 )
213
214 assert resp.status_code == 200, f"Unexpected status {resp.status_code}: {resp.text[:200]}"
215
216 frames = _parse_response_frames(resp.content)
217 assert frames, "Response body must contain at least one msgpack object"
218
219 last = frames[-1]
220 assert last.get("t") == SFRAME_RESULT, (
221 f"Last frame must be RESULT (t='R'). Got: {last}. "
222 f"All frames: {frames}. "
223 "Response must be a msgpack frame stream (P* then R), not a single opaque dict."
224 )
225 assert last.get("ok") is True, f"RESULT frame must have ok=True. Got: {last}"
226
227 for frame in frames[:-1]:
228 t = frame.get("t")
229 assert t == SFRAME_PROGRESS, (
230 f"All frames before the last must be PROGRESS (t='P'). Got t={t!r}: {frame}"
231 )
232
233
234 # ---------------------------------------------------------------------------
235 # T3 — wire_push_stream yields P frames INSIDE the commit loop
236 # ---------------------------------------------------------------------------
237
238 def test_t3_wire_push_stream_emits_progress_during_commit_loop() -> None:
239 """wire_push_stream must yield PROGRESS frames INSIDE the commit insertion
240 loop, not just the single frame emitted before the entire phase starts.
241
242 Without intra-loop heartbeats, CF still times out on large repos even if
243 T1 is fixed: the single P frame before 827 commits is sent immediately,
244 then CF waits 120+ seconds for the next byte while all rows are inserted.
245
246 This test checks the source of musehub_wire.wire_push_stream for a
247 yield _prog(...) call that appears INSIDE the ordered_commits iteration.
248 """
249 from musehub.services import musehub_wire
250
251 source = inspect.getsource(musehub_wire.wire_push_stream)
252
253 # Find the commit iteration loop (enumerate or plain for).
254 loop_marker = "ordered_commits"
255 loop_pos = -1
256 for candidate in ("for _i, wire_commit in enumerate(ordered_commits)", "for wire_commit in ordered_commits"):
257 loop_pos = source.find(candidate)
258 if loop_pos != -1:
259 break
260 assert loop_pos != -1, (
261 "wire_push_stream must iterate over ordered_commits. "
262 "Expected to find 'for wire_commit in ordered_commits' or "
263 "'for _i, wire_commit in enumerate(ordered_commits)'"
264 )
265
266 # Find any yield _prog(...) that appears AFTER the loop start.
267 tail = source[loop_pos:]
268 assert "yield _prog(" in tail, (
269 "wire_push_stream must yield at least one PROGRESS frame INSIDE the "
270 "'for wire_commit in ordered_commits' loop. "
271 "Without intra-loop heartbeats, a push of 800+ commits is silent for "
272 ">120 seconds and CF kills the connection with 524. "
273 "Add: yield _prog(f'committing {i}/{n}…') every ~100 commits."
274 )
File History 1 commit
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 122 days ago