diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index fba3034..8b432fd 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -28,10 +28,19 @@ jobs: - name: Install dependencies run: | uv sync --dev - cd 2-fastapi-r2-streaming && uv run pywrangler dev --help + # Warm every repro directory (uv env, vendored Python packages, npx wrangler) + # so the dev-server startup budget in tests/conftest.py is not spent on installs. + for dir in [0-9]*-*/; do + (cd "$dir" && uv run pywrangler dev --help) + done - name: Run tests run: uv run pytest -vv - name: Run Ruff run: uv run ruff check . + + # The other pre-commit hooks (ruff format, codespell, file hygiene). The `ruff` + # hook is skipped because the step above already runs `ruff check`. + - name: Run pre-commit hooks + run: SKIP=ruff uv run pre-commit run --all-files --show-diff-on-failure diff --git a/2-fastapi-r2-streaming/src/app.py b/2-fastapi-r2-streaming/src/app.py index 3bdd7e2..b5b63a8 100644 --- a/2-fastapi-r2-streaming/src/app.py +++ b/2-fastapi-r2-streaming/src/app.py @@ -48,8 +48,10 @@ async def seed_data(req: Request): r2 = env.BUCKET # Create 128KB of test data — large enough to span many ReadableStream - # chunks (each chunk is typically ~4KB from R2). - data = bytes(range(256)) * 512 # 128KB repeating pattern + # chunks (each chunk is typically ~4KB from R2). Big-endian uint32 word i + # holds i, so no two chunks are alike: a reordered, repeated or dropped + # chunk changes the bytes, not just the length. + data = b"".join(i.to_bytes(4, "big") for i in range(32 * 1024)) # Use the .slice() fix from example 16 to write correctly. js_view = to_js(data) diff --git a/3-httpx-headers/src/entry.py b/3-httpx-headers/src/entry.py index 85d0068..bccbbed 100644 --- a/3-httpx-headers/src/entry.py +++ b/3-httpx-headers/src/entry.py @@ -59,8 +59,10 @@ def pick(received): lower_keys = {k.lower() for k in HEADERS} return {k: v for k, v in received.items() if k.lower() in lower_keys} - return Response.json({ - "headers_sent": HEADERS, - "httpx_received": pick(httpx_received), - "jsfetch_received": pick(jsfetch_received), - }) + return Response.json( + { + "headers_sent": HEADERS, + "httpx_received": pick(httpx_received), + "jsfetch_received": pick(jsfetch_received), + } + ) diff --git a/4-r2-large-binary-roundtrip/pyproject.toml b/4-r2-large-binary-roundtrip/pyproject.toml index 435e92c..51ce3e6 100644 --- a/4-r2-large-binary-roundtrip/pyproject.toml +++ b/4-r2-large-binary-roundtrip/pyproject.toml @@ -5,7 +5,10 @@ description = "Reproduction: large R2 binary round-trip through Python crashes W readme = "README.md" requires-python = ">=3.12" dependencies = [ - "fastapi", + # fastapi 0.142 imports opentelemetry-api, which calls uuid4() at import time. + # Python Workers forbid randomness during startup, and this Worker imports + # FastAPI at module level, so it fails to load. + "fastapi<0.142", ] [dependency-groups] diff --git a/4-r2-large-binary-roundtrip/src/entry.py b/4-r2-large-binary-roundtrip/src/entry.py index 5da638a..7ef35b1 100644 --- a/4-r2-large-binary-roundtrip/src/entry.py +++ b/4-r2-large-binary-roundtrip/src/entry.py @@ -235,29 +235,29 @@ async def compare_paths(key: str, req: Request): total_size = sum(chunk_sizes) first_chunk_size = chunk_sizes[0] if chunk_sizes else 0 - return JSONResponse({ - "key": key, - "r2_body_size": total_size, - "chunk_count": len(parts), - "chunk_sizes": chunk_sizes, - "streaming_would_return": first_chunk_size, - "full_body_would_return": total_size, - "fixed_would_return": total_size, - "ffi_crossings": { - "/asgi-full-body": ( - "JS→Python (getReader) then Python→JS (Response)" - " = 2 crossings" - ), - "/streaming": ( - "JS→Python (getReader) then Python→JS (StreamingResponse)" - " = 2 crossings, but adapter truncates after first yield" - ), - "/fixed": ( - "JS→JS (R2 body → Response)" - " = 0 crossings, data never enters Python" - ), - }, - }) + return JSONResponse( + { + "key": key, + "r2_body_size": total_size, + "chunk_count": len(parts), + "chunk_sizes": chunk_sizes, + "streaming_would_return": first_chunk_size, + "full_body_would_return": total_size, + "fixed_would_return": total_size, + "ffi_crossings": { + "/asgi-full-body": ( + "JS→Python (getReader) then Python→JS (Response) = 2 crossings" + ), + "/streaming": ( + "JS→Python (getReader) then Python→JS (StreamingResponse)" + " = 2 crossings, but adapter truncates after first yield" + ), + "/fixed": ( + "JS→JS (R2 body → Response) = 0 crossings, data never enters Python" + ), + }, + } + ) # -------------------------------------------------------------------------- @@ -327,7 +327,7 @@ async def fetch(self, request): # FIXED path: bypass Python entirely for /fixed/{key} if path.startswith("/fixed/"): - key = path[len("/fixed/"):] + key = path[len("/fixed/") :] if not key: return js.Response.new("Missing key", {"status": 400}) @@ -337,9 +337,11 @@ async def fetch(self, request): # Pass R2 ReadableStream directly to JS Response — data never # enters Python memory. - headers = to_js({ - "content-type": "application/octet-stream", - }) + headers = to_js( + { + "content-type": "application/octet-stream", + } + ) return js.Response.new(obj.body, headers=headers) # All other routes go through FastAPI (ASGI) diff --git a/4-r2-large-binary-roundtrip/uv.lock b/4-r2-large-binary-roundtrip/uv.lock index 893d304..fee9a75 100644 --- a/4-r2-large-binary-roundtrip/uv.lock +++ b/4-r2-large-binary-roundtrip/uv.lock @@ -339,7 +339,7 @@ dev = [ ] [package.metadata] -requires-dist = [{ name = "fastapi" }] +requires-dist = [{ name = "fastapi", specifier = "<0.142" }] [package.metadata.requires-dev] dev = [ diff --git a/LESSONS_LEARNED.md b/LESSONS_LEARNED.md index 2646616..1d10807 100644 --- a/LESSONS_LEARNED.md +++ b/LESSONS_LEARNED.md @@ -127,6 +127,32 @@ Structured post-audit findings from this repository. Each entry is a single wide | **limitations** | The three R2 deployment-only tests were skipped because no deployed Worker URL was supplied. No production deployment was performed. The httpx xfail confirms the still-reproduced header bug; it is not a fix for that platform behavior. The repair did not include a separately documented deliberate-revert negative-control run for every new assertion. | | **takeaway** | **Compare base and head before blaming a tooling PR; record the full tested environment and coverage boundaries.** A green local suite is not evidence for skipped deployment paths. Pinning one tool does not freeze the whole runtime. | +--- + +## 9. General — An Imperative `pytest.xfail()` Is Silent in Both Directions + +| Field | Value | +|-------|-------| +| **category** | test_design | +| **severity** | high | +| **date_discovered** | 2026-09-27 | +| **observation** | Tests called `pytest.xfail()` inside the test body when the bug reproduced, and passed when it did not. After issue 2 was resolved its test still carried the `xfail` branch, so reverting the fix (dropping `workers-runtime-sdk>=1.1.1`) reported `XFAIL` and the run stayed green. For active issue 3, simulating the upstream fix made the test pass, also green, so nothing prompted moving the README entry. | +| **resolution** | Resolved issues have no xfail: a regression fails the run. Active issues use `@pytest.mark.xfail(strict=True, raises=PlatformBugReproduced)`: the bug reproducing is `XFAIL`, the bug disappearing is `XPASS(strict)` (a failure), and any other error (server start, precondition assertions) is a normal failure instead of a silent xfail. | +| **takeaway** | **A known-bug test must go red when its expectation stops holding, in either direction.** This supersedes the "naturally flips to a normal pass" note in entry 6. | + +--- + +## 10. General — Startup Budgets Must Not Depend on Test Order + +| Field | Value | +|-------|-------| +| **category** | flaky_infrastructure | +| **severity** | medium | +| **date_discovered** | 2026-09-27 | +| **observation** | Only `2-fastapi-r2-streaming` got the 300 s CI startup budget; every other directory got 30 s. A cold directory must create its virtualenvs, vendor packages, and fetch wrangler via `npx` before `Ready on`: 28-30 s with warm caches, about 55 s with a cold npm cache. A cold `-k test_3` run failed with `Server failed to start within 30 seconds`. | +| **resolution** | CI warms every directory in its install step (`uv run pywrangler dev --help`, 14-18 s each locally), after which each directory reached `Ready on` in 10-13 s, inside the unchanged 30 s budget. No budget was raised: `2-fastapi-r2-streaming` keeps its existing 300 s under `CI`, and `PYWRANGLER_DEV_TIMEOUT` overrides the budget only when set by hand for a cold local run. The server output is drained on a thread so a silent hang still hits the deadline, and an early exit fails immediately with its exit code. Teardown stops the whole process group (entry 8): terminating only `uv` had left `wrangler` and two `workerd` processes running after every test. | +| **takeaway** | **Pay install costs before the timed wait instead of raising the timeout.** A timeout tuned to whichever test runs first turns cache warmth into test flakiness; warming first removes the order dependence without letting a hang wait longer. | + Reusable rules are in the [shared engineering guidance](https://github.com/adewale/python-workers-examples/blob/main/docs/engineering-guidance.md). The [cross-project retrospective](https://github.com/adewale/python-workers-examples/blob/main/docs/retrospectives/2026-10-09-ruff-rollout.md) diff --git a/README.md b/README.md index 83e75f3..562289d 100644 --- a/README.md +++ b/README.md @@ -15,6 +15,15 @@ Self-contained reproductions of [Python Workers](https://developers.cloudflare.c 3. `uv run pywrangler dev` 4. Press the `b` key to open a browser tab and make a request to the Worker +## Run the tests + +From the repo root, `uv run pytest -vv` starts `pywrangler dev` for each local reproduction and prints a Results summary. + +- **Active issues** are marked `xfail(strict=True)`. The expected outcome is `XFAIL` (the bug still reproduces). If the bug stops reproducing, the test reports `XPASS(strict)` and the run fails, which is the signal to move the issue to Resolved Issues and remove the marker. +- **Resolved issues** are plain regression tests: if the bug comes back, the test fails. +- Each `pywrangler dev` gets 30 s to print `Ready on` (300 s for `2-fastapi-r2-streaming` when `CI` is set). CI warms every directory with `uv run pywrangler dev --help` before the tests, so the budget does not depend on test order. A cold local directory has to vendor packages and fetch wrangler first, so warm it the same way or set `PYWRANGLER_DEV_TIMEOUT=`. +- CI also runs `uv run ruff check .` and `SKIP=ruff uv run pre-commit run --all-files` (ruff format, codespell and file hygiene hooks) in the same job. + ## Active Issues - [**`3-httpx-headers/`**](3-httpx-headers) — The pywrangler-bundled httpx replaces httpcore with a `jsfetch.py` transport that strips the `User-Agent` header to avoid browser CORS preflights. Workers aren't browsers — this causes 403s from APIs like GitHub that require `User-Agent`. **Workaround:** use `js.fetch()` directly. diff --git a/tests/conftest.py b/tests/conftest.py index aea9141..233ec9b 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,5 +1,6 @@ import json import os +import queue import re import signal import socket @@ -16,6 +17,19 @@ REPO_ROOT = Path(__file__).parents[1] +# Seconds to wait for `pywrangler dev` to print "Ready on": 30 s, except 300 s for +# 2-fastapi-r2-streaming under CI, whose first start in CI takes a really long time. +# CI warms every repro directory in its install step so the 30 s budget holds +# whatever the test order. For a cold local run, set PYWRANGLER_DEV_TIMEOUT. +def dev_server_timeout(directory: str) -> float: + value = os.environ.get("PYWRANGLER_DEV_TIMEOUT") + if value: + return float(value) + if "CI" in os.environ and directory.startswith("2-"): + return 300.0 + return 30.0 + + def pytest_terminal_summary(terminalreporter, exitstatus, config): """Print a clear summary of which platform bugs were confirmed.""" xfailed = terminalreporter.stats.get("xfailed", []) @@ -32,7 +46,13 @@ def pytest_terminal_summary(terminalreporter, exitstatus, config): terminalreporter.line(f" PASSED {name}") for report in failed: name = report.nodeid.split("::")[-1] - terminalreporter.line(f" FAILED {name}") + if "XPASS(strict)" in str(report.longrepr): + terminalreporter.line( + f" NOT REPRODUCED {name}: the platform bug may be fixed upstream; " + "move it to Resolved Issues and remove its xfail marker" + ) + else: + terminalreporter.line(f" FAILED {name}") def pytest_addoption(parser): @@ -106,6 +126,9 @@ def pywrangler_dev_server(directory: str, echo_url: str | None = None): command = ["uv", "run", "pywrangler", "dev", "--port", str(port)] if echo_url: command.extend(["--var", f"ECHO_URL:{echo_url}"]) + # `uv run pywrangler dev` forks npx -> wrangler -> workerd. Start it in its own + # process group so teardown can stop the whole tree; terminating only `uv` + # leaves wrangler and workerd running after the test session ends. process = subprocess.Popen( command, cwd=REPO_ROOT / directory, @@ -115,29 +138,45 @@ def pywrangler_dev_server(directory: str, echo_url: str | None = None): start_new_session=True, ) - # Wait for server to be ready - ready = False - timeout = 30 - if "CI" in os.environ and directory.startswith("2-"): - # Starting the server the first time takes a really long time in CI. - timeout = 300 + # Drain the server's output on a thread for its whole lifetime, so a silent hang + # still hits the timeout and a chatty server never blocks on a full pipe. + lines: queue.Queue = queue.Queue() + + def _pump(): + for line in process.stdout: + lines.put(line) + lines.put(None) - start_time = time.time() + Thread(target=_pump, daemon=True).start() - while not ready and time.time() - start_time < timeout: - line = process.stdout.readline() - if not line and process.poll() is not None: + timeout = dev_server_timeout(directory) + deadline = time.monotonic() + timeout + ready = False + while not ready: + remaining = deadline - time.monotonic() + if remaining <= 0: + break + try: + line = lines.get(timeout=remaining) + except queue.Empty: break - if line: - print(line.rstrip(), file=sys.stdout) # Also print to stdout - if "[wrangler:info] Ready on" in line: - ready = True - break - time.sleep(0.1) + if line is None: + _signal_server(process, signal.SIGTERM) + process.wait() + raise RuntimeError( + f"pywrangler dev in {directory} exited with code " + f"{process.returncode} before it was ready" + ) + print(line.rstrip(), file=sys.stdout) # Also print to stdout + if "[wrangler:info] Ready on" in line: + ready = True if not ready: _signal_server(process, signal.SIGTERM) - raise RuntimeError(f"Server failed to start within {timeout} seconds") + raise RuntimeError( + f"Server in {directory} failed to start within {timeout:g} seconds " + "(set PYWRANGLER_DEV_TIMEOUT to change the budget)" + ) try: yield port @@ -153,9 +192,9 @@ def pywrangler_dev_server(directory: str, echo_url: str | None = None): @pytest.fixture def dev_server(request, header_echo_server): """Fixture that starts a dev server for the appropriate directory based on test name.""" - if request.node.get_closest_marker("skip") or request.node.get_closest_marker( - "xfail" - ): + # Only an explicit skip avoids starting the server. Active-issue tests are + # marked xfail(strict=True) and must still run against a real server. + if request.node.get_closest_marker("skip"): yield return diff --git a/tests/test_examples.py b/tests/test_examples.py index c8c0d93..c1af719 100644 --- a/tests/test_examples.py +++ b/tests/test_examples.py @@ -4,6 +4,27 @@ EXPECTED_128KB = 131072 +class PlatformBugReproduced(Exception): + """Raised by a repro test when it observes the known platform bug. + + Tests for active issues are marked ``xfail(strict=True, + raises=PlatformBugReproduced)``. That makes the outcome loud in both + directions: + + - the bug reproduces -> xfailed (CONFIRMED in the Results summary); + - the bug no longer reproduces -> XPASS(strict), which fails the run, the + signal to move the issue to "Resolved Issues" and drop the marker; + - anything else goes wrong (server, precondition assertions) -> a normal + failure, never a silent xfail. + + Tests for resolved issues have no marker, so a regression fails hard. + """ + + +def active_issue(reason): + return pytest.mark.xfail(strict=True, raises=PlatformBugReproduced, reason=reason) + + def test_2_fastapi_r2_streaming(dev_server): port = dev_server @@ -16,6 +37,12 @@ def test_2_fastapi_r2_streaming(dev_server): read_resp = requests.get(f"http://localhost:{port}/read/test-file") assert read_resp.status_code == 200 assert len(read_resp.content) == EXPECTED_128KB + # POST /seed stores 32768 big-endian uint32 words where word i holds i. + # A reordered, repeated or zeroed chunk keeps the length, not the words. + body = read_resp.content + words = [int.from_bytes(body[i : i + 4], "big") for i in range(0, len(body), 4)] + misplaced = next((i for i, word in enumerate(words) if word != i), None) + assert misplaced is None, f"word {misplaced} holds {words[misplaced]}" # Compare endpoint — validates we can read all chunks compare_resp = requests.get(f"http://localhost:{port}/compare/test-file") @@ -24,20 +51,26 @@ def test_2_fastapi_r2_streaming(dev_server): assert compare["full_body_size"] == EXPECTED_128KB assert compare["chunk_count"] > 1 - # StreamingResponse path — should return all data, but platform bug - # causes it to return only the first chunk. + # StreamingResponse path — resolved upstream by workers-runtime-sdk>=1.1.1. + # This is now a regression test: truncation must fail the run. stream_resp = requests.get(f"http://localhost:{port}/stream/test-file") assert stream_resp.status_code == 200 streamed_size = len(stream_resp.content) - - if streamed_size < EXPECTED_128KB: - pytest.xfail( - f"Platform bug confirmed: StreamingResponse returned {streamed_size} bytes, " - f"expected {EXPECTED_128KB}. ASGI adapter truncates to first chunk." - ) - assert streamed_size == EXPECTED_128KB + assert streamed_size == EXPECTED_128KB, ( + f"Regression of resolved issue 2: StreamingResponse returned {streamed_size} " + f"bytes, expected {EXPECTED_128KB} (ASGI adapter truncated to the first chunk)." + ) + words = [ + int.from_bytes(stream_resp.content[i : i + 4], "big") + for i in range(0, len(stream_resp.content), 4) + ] + misplaced = next((i for i, word in enumerate(words) if word != i), None) + assert misplaced is None, f"stream word {misplaced} holds {words[misplaced]}" +@active_issue( + "Issue 3: pywrangler-bundled httpx strips User-Agent (jsfetch.py HEADERS_TO_IGNORE)" +) def test_3_httpx_headers(dev_server, header_echo_server): port = dev_server response = requests.get(f"http://localhost:{port}/test") @@ -60,7 +93,7 @@ def test_3_httpx_headers(dev_server, header_echo_server): assert httpx_headers.get("X-Custom") == "preserved" if "User-Agent" not in httpx_headers: - pytest.xfail( + raise PlatformBugReproduced( "Platform bug confirmed: httpx User-Agent header was stripped by " "jsfetch.py HEADERS_TO_IGNORE." ) @@ -90,6 +123,9 @@ def test_5_sync_http_libraries(dev_server, header_echo_server): ) +@active_issue( + "Issue 4, bug 1: ASGI adapter truncates StreamingResponse to the first chunk" +) def test_4a_streaming_truncation(deployed_url): """Bug 1: ASGI adapter truncates StreamingResponse to first chunk.""" base = deployed_url @@ -141,7 +177,7 @@ def test_4a_streaming_truncation(deployed_url): print(" chunk from the async generator and dropped the rest.") print(" StreamingResponse is broken for any file that spans") print(" multiple R2 chunks (~4KB each).") - pytest.xfail( + raise PlatformBugReproduced( f"Bug 1: StreamingResponse returned {streamed_size} bytes, " f"expected {expected_bytes}. ASGI adapter truncates async " f"generators to the first yielded chunk." @@ -151,13 +187,17 @@ def test_4a_streaming_truncation(deployed_url): assert streamed_size == expected_bytes +@active_issue( + "Issue 4, bug 2: Wasm memory exhausted by FFI round-trip of large R2 objects" +) def test_4b_memory_crash_probe(deployed_url): """Bug 2: Find the Wasm memory crash threshold for this Worker. Seeds and round-trips in separate requests (fresh isolate per step) - at escalating sizes. A minimal Worker (FastAPI only) may not crash - even at 100MB. A production Worker with many packages crashes at - ~42MB. + at escalating sizes. A production Worker with many packages crashes at + ~42MB; this repro Worker has crashed at 50MB. If it survives every + step up to 100MB the run fails (XPASS strict): check whether the + platform changed before moving issue 4 to Resolved Issues. """ base = deployed_url step_mb = 10 @@ -178,9 +218,7 @@ def test_4b_memory_crash_probe(deployed_url): # Seed in its own request print(f"\n {size_mb}MB: seeding...", end="", flush=True) - seed_resp = requests.post( - f"{base}/probe/seed/{size_mb}", timeout=60 - ) + seed_resp = requests.post(f"{base}/probe/seed/{size_mb}", timeout=60) if seed_resp.status_code >= 500: print(f" CRASHED (HTTP {seed_resp.status_code})") crash_at = size_mb @@ -192,9 +230,7 @@ def test_4b_memory_crash_probe(deployed_url): # Round-trip in its own request (fresh isolate) try: - rt_resp = requests.get( - f"{base}/probe/roundtrip/{size_mb}", timeout=60 - ) + rt_resp = requests.get(f"{base}/probe/roundtrip/{size_mb}", timeout=60) except requests.exceptions.ConnectionError: print(" CRASHED (connection reset)") crash_at = size_mb @@ -221,7 +257,7 @@ def test_4b_memory_crash_probe(deployed_url): print(f" At {crash_at}MB, Wasm linear memory cannot hold 3 copies") print(f" ({crash_at * 3}MB total: R2 buffer + Python bytes + JS Response).") print(" Workers with more packages crash at lower sizes.") - pytest.xfail( + raise PlatformBugReproduced( f"Bug 2: FFI round-trip crashed at {crash_at}MB " f"(last success: {last_ok}MB). Wasm memory exhausted by " f"3x copies of R2 data crossing the FFI boundary." @@ -262,12 +298,15 @@ def test_4c_diagnostics(deployed_url): print(f" Chunk count: {diag['chunk_count']}") print(f" Chunk sizes: {diag['chunk_sizes']}") print("\n What each endpoint would return:") - print(f" /streaming/ {diag['streaming_would_return']:,} bytes" - f" (first chunk only — Bug 1)") - print(f" /asgi-full-body/ {diag['full_body_would_return']:,} bytes" - f" (all chunks joined)") - print(f" /fixed/ {diag['fixed_would_return']:,} bytes" - f" (JS bypass)") + print( + f" /streaming/ {diag['streaming_would_return']:,} bytes" + f" (first chunk only — Bug 1)" + ) + print( + f" /asgi-full-body/ {diag['full_body_would_return']:,} bytes" + f" (all chunks joined)" + ) + print(f" /fixed/ {diag['fixed_would_return']:,} bytes (JS bypass)") print("\n FFI boundary crossings:") for path, desc in diag["ffi_crossings"].items(): print(f" {path}: {desc}")