From 8281baa9fc0fa2b2ba3944c61c7b24d233d86152 Mon Sep 17 00:00:00 2001 From: Build System Date: Mon, 18 May 2026 12:51:52 +0200 Subject: [PATCH 1/3] fix: zstd truncation/abort on proxy_buffering off (bug B) + regression tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Production (deb.myguard.nl, `proxy_buffering off`) served truncated / aborted zstd responses; nginx logged "zero size buf in writer". Root cause, two coupled defects in ngx_http_zstd_filter_compress(): 1. The compress directive was selected from ctx->action only; ctx->flush never mapped to ZSTD_e_flush. The COMPRESS->FLUSH action transition only happens after a call that returned rc > 0. Under proxy_buffering off the upstream forces a flush around a chunk that libzstd consumes/buffers internally with rc == 0, so the directive stayed ZSTD_e_continue and libzstd was never told to flush. 2. The empty-buffer emit guard had a `!(rc == 0 && ctx->flush)` exception that *forwarded a zero-size buffer* for such a content-less flush. nginx's ngx_http_write_filter rejects a zero-size buffer ("zero size buf in writer") and aborts the request — the truncation/empty-response users saw. Fix (both parts are required; each alone is wrong): - Directive: ZSTD_e_flush when (action == FLUSH || ctx->flush), so a pending flush forces libzstd to disgorge buffered output (mirrors the stock nginx gzip/brotli sync-flush behaviour). - Empty-buffer guard: never forward a zero-size non-terminal buffer. When it is a content-less *completed* flush (rc == 0 && ctx->flush: per the libzstd contract rc == 0 from a flush means fully drained), clear NGX_HTTP_GZIP_BUFFERED and ctx->flush and emit nothing. Clearing ctx->flush is essential for loop termination: the body filter's inner loop stays alive while ctx->flush is set (ngx_http_zstd_filter_add_data keeps returning NGX_OK); a naive "suppress the empty buffer but leave ctx->flush set" change spins a worker at 100% CPU (NGX_AGAIN livelock — observed and rejected). Regression tests (decode-and-diff, the discipline the prior misses lacked): - tools/test_proxy_unbuffered_truncation.py — the focused bug-B repro: proxy_buffering off + chunked no-CL upstream, boundary size matrix around ZSTD_CStreamInSize, byte-exact full-body compare. - tools/test_compression_matrix.py — broad ~504-cell matrix: {proxy buffered, proxy unbuffered, static} x {incompressible, highly-compressible, css} x boundary sizes x {zstd, gzip+zstd, gzip-only, identity}; every cell decode-and-diffs or asserts correct fallback. Both wired into build-test.yml `tests` and `tests-asan`, with step timeouts so a livelock is a hard FAIL. Verification: focused test 35/35 byte-exact on the fix and FAILS on clean master (35x "zero size buf in writer") — Principle-0 fail-first satisfied. Broad matrix 504/504 green on the fix. Under ASAN+UBSAN the fix serves valid zstd; the only LeakSanitizer report is an identical pre-existing artifact of SIGKILLing a master_process-off nginx (byte -for-byte the same on clean master), not introduced here. Refs the recurring truncation/terminal-frame class: a209f96, 2af5889, PR#23/#49, #36. --- .github/workflows/build-test.yml | 41 +++ filter/ngx_http_zstd_filter_module.c | 66 +++- tools/test_compression_matrix.py | 384 ++++++++++++++++++++++ tools/test_proxy_unbuffered_truncation.py | 341 +++++++++++++++++++ 4 files changed, 827 insertions(+), 5 deletions(-) create mode 100644 tools/test_compression_matrix.py create mode 100644 tools/test_proxy_unbuffered_truncation.py diff --git a/.github/workflows/build-test.yml b/.github/workflows/build-test.yml index 467a97e..42d3c1a 100644 --- a/.github/workflows/build-test.yml +++ b/.github/workflows/build-test.yml @@ -484,6 +484,38 @@ jobs: python3 tools/test_terminal_frame.py --nginx-binary "$TEST_NGINX_BINARY" --port 18085 echo "✓ Terminal-frame regression test passed" + - name: Run proxy_buffering-off truncation regression (bug B) + run: | + cd "$GITHUB_WORKSPACE" + # Reproduces the production trigger: zstd over an unbuffered + # (proxy_buffering off) chunked, no-Content-Length upstream. + # Decodes and byte-compares the full body across the + # ZSTD_CStreamInSize buffer-boundary size matrix. Fails on a + # module that forwards a zero-size buffer ("zero size buf in + # writer" -> aborted/truncated response). + python3 tools/test_proxy_unbuffered_truncation.py \ + --nginx-binary "$TEST_NGINX_BINARY" \ + --port 18086 --backend-port 18087 + echo "✓ proxy_buffering-off truncation regression passed" + + - name: Run broad compression correctness matrix + timeout-minutes: 12 + run: | + cd "$GITHUB_WORKSPACE" + # ~504 cells: {proxy-buffered, proxy-unbuffered, static} + # x {incompressible, highly-compressible, css} + # x {1, CStreamInSize +/-1, 200000, 1MiB} + # x {zstd, gzip+zstd, gzip-only, identity} x repeat. + # Every cell decodes per Content-Encoding and byte-compares + # to origin, or asserts correct fallback. Catches the whole + # truncation / zero-size-buf / livelock bug class across the + # circumstances it has historically recurred in (a step-level + # timeout makes a CPU-spin livelock a hard FAIL, not a hang). + python3 tools/test_compression_matrix.py \ + --nginx-binary "$TEST_NGINX_BINARY" \ + --port 18096 --backend-port 18097 + echo "✓ compression correctness matrix passed" + - name: Test summary if: success() run: echo "✓ All Perl + Python end-to-end tests passed" @@ -535,6 +567,15 @@ jobs: python3 tools/test_encoding.py --nginx-binary "$TEST_NGINX_BINARY" --port 18091 --fixture-lines 1900 python3 tools/test_encoding.py --nginx-binary "$TEST_NGINX_BINARY" --port 18092 --concurrent-requests 2 python3 tools/test_terminal_frame.py --nginx-binary "$TEST_NGINX_BINARY" --port 18093 + python3 tools/test_proxy_unbuffered_truncation.py \ + --nginx-binary "$TEST_NGINX_BINARY" --port 18094 --backend-port 18095 + # Matrix under ASAN/UBSAN at repeat=1 (252 cells) — the + # truncation/zero-size-buf/livelock class has UB+lifetime + # history, so exercise the full circumstance matrix with + # sanitizers, not just the happy path. + python3 tools/test_compression_matrix.py \ + --nginx-binary "$TEST_NGINX_BINARY" --port 18096 \ + --backend-port 18097 --repeat 1 echo "✓ All smoke tests clean under ASAN+UBSAN" - name: zstd_dict_file reload leak check (CDict lifecycle) diff --git a/filter/ngx_http_zstd_filter_module.c b/filter/ngx_http_zstd_filter_module.c index 9174107..f146912 100644 --- a/filter/ngx_http_zstd_filter_module.c +++ b/filter/ngx_http_zstd_filter_module.c @@ -573,10 +573,24 @@ ngx_http_zstd_filter_compress(ngx_http_request_t *r, ngx_http_zstd_ctx_t *ctx) pos_in = ctx->buffer_in.pos; pos_out = ctx->buffer_out.pos; - /* Determine the compression directive based on action state */ + /* + * Determine the compression directive. + * + * END always wins (terminal frame). Otherwise a pending flush + * (ctx->flush) must map to ZSTD_e_flush even while the action state + * machine is still COMPRESS: that machine only transitions + * COMPRESS->FLUSH *after* a call that returned rc > 0 (zstd already + * had output to drain). Under `proxy_buffering off` the upstream + * forces a flush around a chunk that zstd consumes/buffers + * internally with rc == 0; if the directive stayed ZSTD_e_continue + * there, libzstd is never told to flush and holds those bytes + * indefinitely. Mapping ctx->flush -> ZSTD_e_flush forces libzstd + * to disgorge whatever it has buffered, exactly as the stock nginx + * gzip/brotli body filters issue a sync flush on a pending flush. + */ if (ctx->action == NGX_HTTP_ZSTD_FILTER_END) { directive = ZSTD_e_end; - } else if (ctx->action == NGX_HTTP_ZSTD_FILTER_FLUSH) { + } else if (ctx->action == NGX_HTTP_ZSTD_FILTER_FLUSH || ctx->flush) { directive = ZSTD_e_flush; } else { directive = ZSTD_e_continue; @@ -687,9 +701,51 @@ ngx_http_zstd_filter_compress(ngx_http_request_t *r, ngx_http_zstd_ctx_t *ctx) (size_t) ngx_buf_size(ctx->out_buf), rc == 0, last, ctx->last, ctx->flush, (ngx_uint_t) ctx->action); - if (ngx_buf_size(ctx->out_buf) == 0 && !last && !(rc == 0 && ctx->flush)) { - ngx_log_debug0(NGX_LOG_DEBUG_HTTP, r->connection->log, 0, - "zstd emit: suppressed empty non-terminal buffer"); + /* + * Empty, non-terminal output buffer handling. + * + * Two distinct empty-buffer cases reach here: + * + * 1. A content-less *completed flush*: ctx->flush is set and the + * ZSTD_e_flush call returned rc == 0 with zero bytes produced. + * Per the libzstd contract rc == 0 from a flush means the + * encoder is fully drained — there is genuinely nothing to + * send. The OLD code's `!(rc == 0 && ctx->flush)` exception + * forwarded a zero-size buffer here, which nginx's + * ngx_http_write_filter rejects ("zero size buf in writer") + * and aborts the request — bug B, under `proxy_buffering off` + * where the upstream forces such flushes. + * + * The flush is COMPLETE, so satisfy and clear it here: drop the + * NGX_HTTP_GZIP_BUFFERED bit and ctx->flush, emit nothing, and + * return NGX_AGAIN. Clearing ctx->flush is what makes the body + * filter's inner loop terminate: ngx_http_zstd_filter_add_data() + * keeps the loop alive while ctx->flush is set (expecting this + * function to consume it); a naive "suppress the empty buffer + * but leave ctx->flush set" change spins the worker at 100% CPU + * (a NGX_AGAIN livelock — observed). Clearing it lets the next + * add_data() fall through to NGX_DECLINED (input drained) and + * break the loop cleanly. + * + * 2. Any other empty, non-terminal buffer (no pending flush): + * nothing to send and nothing to satisfy — just suppress and + * return NGX_AGAIN as before. + * + * The genuine terminal empty buffer (last) is never suppressed; it + * falls through to be emitted with last_buf set below. + */ + if (ngx_buf_size(ctx->out_buf) == 0 && !last) { + if (rc == 0 && ctx->flush) { + r->connection->buffered &= ~NGX_HTTP_GZIP_BUFFERED; + ctx->flush = 0; + ngx_log_debug0(NGX_LOG_DEBUG_HTTP, r->connection->log, 0, + "zstd emit: content-less flush completed, " + "cleared (no empty buffer forwarded)"); + } else { + ngx_log_debug0(NGX_LOG_DEBUG_HTTP, r->connection->log, 0, + "zstd emit: suppressed empty non-terminal " + "buffer"); + } return NGX_AGAIN; } diff --git a/tools/test_compression_matrix.py b/tools/test_compression_matrix.py new file mode 100644 index 0000000..808deb7 --- /dev/null +++ b/tools/test_compression_matrix.py @@ -0,0 +1,384 @@ +#!/usr/bin/env python3 +"""Broad correctness matrix for the zstd filter. + +The module's truncation / zero-size-buffer / terminal-frame bug class +(a209f96, 2af5889, PR#23/#49, bug A #36, bug B) keeps recurring in +*different* circumstances, and each prior regression test pinned only +the one slice that had just broken. This test crosses the axes the bug +class has actually hit and asserts the right outcome in **every cell**: + + * transport : proxy (buffering on), proxy (buffering off), static + * payload : incompressible, highly-compressible, real-CSS-like + * size : 1, 1024, CStreamInSize-1/=/+1, 200000, 1 MiB + * Accept-Encoding: "zstd" -> must be zstd, byte-exact + "gzip, br, zstd" -> some C-E set, body decodes to + origin (zstd OR a co-filter + wins; either is correct as + long as it round-trips) + "gzip" -> must NOT be zstd; body intact + "" (identity) -> no C-E; body intact + +Every cell verifies the client can reconstruct the exact origin bytes +(decode per the returned Content-Encoding, then byte-compare) — never +"got 200" or "got non-empty". A truncation/abort shows up as a short +body, a decode error, or a connection failure. + +Rig discipline (learned the hard way): + * threaded backend (single-thread is OOM/stall-killed under load) + * backend self-check before trusting any end-to-end result + * distinct bytes per (payload,size) so a wrong-body is detectable + * fresh request each cell; HTTP error => failure, not silent reuse + +Self-contained: stdlib + the ``zstd`` and ``gzip`` CLIs (+ ``brotli`` +if present; brotli cells are skipped when it is not). No PHP-FPM. +""" +from __future__ import annotations + +import argparse +import gzip as _gzip +import http.server +import itertools +import pathlib +import shutil +import socket +import socketserver +import subprocess +import sys +import tempfile +import threading +import time +import urllib.request + +CSTREAM_IN = 131072 # ZSTD_CStreamInSize on libzstd 1.5.x — the + # historical boundary for this bug class. +SIZES = [1, 1024, CSTREAM_IN - 1, CSTREAM_IN, CSTREAM_IN + 1, + 200000, 1024 * 1024] +PAYLOADS = ("incompressible", "compressible", "css") + + +def parse_args() -> argparse.Namespace: + p = argparse.ArgumentParser( + description="Broad zstd filter correctness matrix.") + p.add_argument("--nginx-binary", required=True) + p.add_argument("--filter-module") + p.add_argument("--static-module") + p.add_argument("--zstd-bin", default="zstd") + p.add_argument("--port", type=int, default=18096) + p.add_argument("--backend-port", type=int, default=18097) + p.add_argument("--repeat", type=int, default=2, + help="Re-request each cell N times (intermittent " + "bugs surface on repeats).") + return p.parse_args() + + +def detect(explicit, nginx: pathlib.Path, name: str): + if explicit: + return pathlib.Path(explicit) + sib = nginx.parent / name + return sib if sib.exists() else None + + +def make_payload(kind: str, n: int) -> bytes: + if kind == "incompressible": + buf = bytearray(n) + x = (n * 2654435761) & 0xFFFFFFFF + for i in range(n): + x = (x * 1103515245 + 12345) & 0xFFFFFFFF + buf[i] = (x >> 16) & 0xFF + return bytes(buf) + if kind == "compressible": + # Highly compressible: zstd buffers internally and produces + # zero output on early/forced-flush calls — the exact bug-B + # path. A short unique prefix keeps (kind,size) bodies distinct. + head = f"CELL-{kind}-{n}-".encode() + return (head + b"A" * max(0, n - len(head)))[:n] + # "css": realistic mid-ratio text, distinct per size. + unit = (f"/* cell {n} */.c{n}" + "{color:#abcdef;margin:0 auto;padding:1px}\n").encode() + body = (unit * (n // len(unit) + 1))[:n] + return body + + +class _Handler(http.server.BaseHTTPRequestHandler): + """GET /p// -> exactly bytes of , **chunked, + no Content-Length**, streamed in small records.""" + protocol_version = "HTTP/1.1" + fixtures: dict = {} + + def log_message(self, *a): + pass + + def do_GET(self): + parts = self.path.split("?")[0].strip("/").split("/") + if len(parts) != 3 or parts[0] != "p": + self.send_response(404) + self.send_header("Content-Length", "0") + self.end_headers() + return + key = (parts[1], int(parts[2])) + data = self.fixtures.get(key) + if data is None: + self.send_response(404) + self.send_header("Content-Length", "0") + self.end_headers() + return + self.send_response(200) + self.send_header("Content-Type", "text/css") + self.send_header("Transfer-Encoding", "chunked") + self.end_headers() + mv = memoryview(data) + for i in range(0, len(mv), 16384): + c = bytes(mv[i:i + 16384]) + self.wfile.write(b"%X\r\n" % len(c) + c + b"\r\n") + self.wfile.write(b"0\r\n\r\n") + + +class _Srv(socketserver.ThreadingMixIn, socketserver.TCPServer): + allow_reuse_address = True + daemon_threads = True + + +def wait_port(port: int, timeout: float = 10.0) -> None: + end = time.time() + timeout + while time.time() < end: + try: + with socket.create_connection(("127.0.0.1", port), 0.5): + return + except OSError: + time.sleep(0.1) + raise RuntimeError(f"nothing on 127.0.0.1:{port}") + + +def fetch(port: int, path: str, accept_encoding: str): + hdrs = {"User-Agent": "zstd-matrix/1.0"} + if accept_encoding: + hdrs["Accept-Encoding"] = accept_encoding + req = urllib.request.Request(f"http://127.0.0.1:{port}{path}", + headers=hdrs) + with urllib.request.urlopen(req, timeout=25) as resp: + return resp.read(), (resp.headers.get("Content-Encoding") + or "").lower() + + +def decode(zstd_bin: str, blob: bytes, enc: str) -> bytes: + if enc == "zstd": + if blob[:4] != b"\x28\xb5\x2f\xfd": + raise RuntimeError("zstd C-E but no zstd magic " + f"(hex={blob[:8].hex()})") + r = subprocess.run([zstd_bin, "-dq", "-c"], input=blob, + capture_output=True) + if r.returncode != 0: + raise RuntimeError("zstd -d failed (premature end): " + + r.stderr.decode("utf-8", "replace")) + return r.stdout + if enc == "gzip": + return _gzip.decompress(blob) + if enc == "br": + br = shutil.which("brotli") + if not br: + raise RuntimeError("br C-E but no brotli CLI to verify") + r = subprocess.run([br, "-dc"], input=blob, capture_output=True) + if r.returncode != 0: + raise RuntimeError("brotli -d failed") + return r.stdout + if enc in ("", "identity"): + return blob + raise RuntimeError(f"unexpected Content-Encoding {enc!r}") + + +def main() -> int: + args = parse_args() + nginx = pathlib.Path(args.nginx_binary) + if not nginx.exists(): + raise FileNotFoundError(nginx) + mods = [m for m in (detect(args.filter_module, nginx, + "ngx_http_zstd_filter_module.so"), + detect(args.static_module, nginx, + "ngx_http_zstd_static_module.so")) + if m] + have_brotli = shutil.which("brotli") is not None + + with tempfile.TemporaryDirectory(prefix="zstd-matrix-") as td: + root = pathlib.Path(td) + sroot = root / "static" + logs = root / "logs" + sroot.mkdir() + logs.mkdir() + + fixtures: dict = {} + for kind, n in itertools.product(PAYLOADS, SIZES): + data = make_payload(kind, n) + fixtures[(kind, n)] = data + (sroot / f"{kind}-{n}").write_bytes(data) + _Handler.fixtures = fixtures + + backend = _Srv(("127.0.0.1", args.backend_port), _Handler) + threading.Thread(target=backend.serve_forever, + daemon=True).start() + try: + wait_port(args.backend_port) + # backend self-check (rig discipline) + for kind, n in (("incompressible", 1), + ("compressible", CSTREAM_IN), + ("css", 1024 * 1024)): + raw = urllib.request.urlopen( + f"http://127.0.0.1:{args.backend_port}/p/{kind}/{n}", + timeout=10).read() + if raw != fixtures[(kind, n)]: + raise RuntimeError( + f"backend self-check failed for {kind}/{n}") + + load = "".join(f"load_module {m};\n" for m in mods) + conf = root / "nginx.conf" + conf.write_text(f"""{load}worker_processes 1; +error_log {logs}/error.log warn; +pid {root}/nginx.pid; +events {{ worker_connections 256; }} +http {{ + access_log off; + default_type text/css; + zstd on; + zstd_comp_level 6; + zstd_min_length 1; + zstd_types text/css; + gzip on; + gzip_comp_level 5; + gzip_min_length 1; + gzip_types text/css; + server {{ + listen 127.0.0.1:{args.port}; + # transport: proxy, buffering ON (default) + location /bufon/ {{ + proxy_pass http://127.0.0.1:{args.backend_port}/p/; + proxy_http_version 1.1; + proxy_set_header Connection ""; + }} + # transport: proxy, buffering OFF (the bug-B production shape) + location /bufoff/ {{ + proxy_pass http://127.0.0.1:{args.backend_port}/p/; + proxy_http_version 1.1; + proxy_set_header Connection ""; + proxy_buffering off; + }} + # transport: static file (sendfile path) + location /static/ {{ + alias {sroot}/; + }} + }} +}} +""", encoding="utf-8") + + proc = subprocess.Popen( + [str(nginx), "-p", str(root), "-c", str(conf), + "-g", "daemon off; master_process off;"], + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, + text=True) + try: + wait_port(args.port) + ae_axis = ["zstd", "gzip", ""] + if have_brotli: + ae_axis.append("gzip, br, zstd") + else: + ae_axis.append("gzip, zstd") # zstd should win + + transports = { + "bufon": lambda k, n: f"/bufon/{k}/{n}", + "bufoff": lambda k, n: f"/bufoff/{k}/{n}", + "static": lambda k, n: f"/static/{k}-{n}", + } + + failures: list[str] = [] + cells = 0 + for tname, mkpath in transports.items(): + for kind in PAYLOADS: + for n in SIZES: + origin = fixtures[(kind, n)] + for ae in ae_axis: + for attempt in range(args.repeat): + cells += 1 + lbl = (f"{tname} {kind} n={n} " + f"AE={ae!r} #{attempt+1}") + try: + body, enc = fetch( + args.port, + mkpath(kind, n), ae) + except Exception as e: # noqa + failures.append( + f"{lbl}: request failed: {e}") + continue + # Correctness rules: + # AE 'zstd' -> must be zstd. + # AE 'gzip' -> must NOT be zstd. + # any -> body decodes to origin. + if ae == "zstd" and enc != "zstd": + failures.append( + f"{lbl}: expected zstd, " + f"got C-E={enc!r}") + continue + if (ae == "gzip" + and enc == "zstd"): + failures.append( + f"{lbl}: zstd served to " + f"gzip-only client") + continue + try: + dec = decode(args.zstd_bin, + body, enc) + except Exception as e: # noqa + failures.append( + f"{lbl}: decode failed " + f"(C-E={enc!r}): {e}") + continue + if dec != origin: + failures.append( + f"{lbl}: body mismatch " + f"(C-E={enc!r} got " + f"{len(dec)}B want {n}B)") + + if failures: + sys.stderr.write( + f"MATRIX FAILED: {len(failures)}/{cells} " + f"cells:\n") + for f in failures[:30]: + sys.stderr.write(f" - {f}\n") + if len(failures) > 30: + sys.stderr.write( + f" ... +{len(failures) - 30} more\n") + el = logs / "error.log" + if el.exists(): + z = [ln for ln in el.read_text( + "utf-8", "replace").splitlines() + if "zero size buf in writer" in ln] + if z: + sys.stderr.write( + f" nginx logged {len(z)} " + f"'zero size buf in writer'\n") + return 1 + + print(f"OK: {cells} matrix cells " + f"({len(transports)} transports x " + f"{len(PAYLOADS)} payloads x {len(SIZES)} sizes " + f"x {len(ae_axis)} AE x {args.repeat}) " + f"all round-tripped byte-exact / correct " + f"fallback" + + ("" if have_brotli + else " [brotli CLI absent: br-axis skipped]")) + return 0 + finally: + proc.terminate() + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait(timeout=5) + finally: + backend.shutdown() + backend.server_close() + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except Exception as exc: # noqa: BLE001 + print(f"ERROR: {exc}", file=sys.stderr) + raise diff --git a/tools/test_proxy_unbuffered_truncation.py b/tools/test_proxy_unbuffered_truncation.py new file mode 100644 index 0000000..b5b1b23 --- /dev/null +++ b/tools/test_proxy_unbuffered_truncation.py @@ -0,0 +1,341 @@ +#!/usr/bin/env python3 +"""Regression test for bug B: silent truncation / aborted response when +zstd compresses a proxied response with ``proxy_buffering off``. + +Background +---------- +Production (deb.myguard.nl, ``proxy_buffering off`` in the WordPress +vhost) served truncated / empty zstd responses. Root cause: under +``proxy_buffering off`` the upstream forces a flush around body data on +a compress call that feeds libzstd nothing (or sub-block-size input it +buffers internally); the filter then forwarded a *zero-size* buffer, +which nginx's ``ngx_http_write_filter`` rejects with +``zero size buf in writer`` and aborts the request. The body filter's +compress-directive selection ignored ``ctx->flush`` and the empty-buffer +emit guard had a ``!(rc == 0 && ctx->flush)`` exception that let the +empty buffer through. + +Why the existing suite missed it +-------------------------------- +``t/00-filter.t`` chunked cases use ``--- ignore_response`` (assert a +log line, never decode the body). ``test_terminal_frame.py`` uses a +tiny body that never crosses an output buffer. None used +``proxy_buffering off``. This test closes that gap: it reproduces the +exact production transport (proxy_pass -> chunked, no Content-Length, +``proxy_buffering off``) and **decodes + byte-compares the full body** +across the buffer-boundary size matrix. + +It MUST fail on a module without the fix (truncated/aborted -> decoded +bytes differ or curl fails) and pass with it. + +Hard test-rig discipline (learned the hard way during the bug-B hunt) +-------------------------------------------------------------------- +* Decode and byte-compare the *whole* body and assert exact length — + never trust an HTTP 200 or a non-empty body alone. +* Assert the response for size N actually has length N: guards against + a stale / wrong-body false-green. +* Threaded backend: a single-threaded http.server is OOM/stall-killed + under the boundary matrix and produces mass false failures. +* Verify the backend answers correctly *before* starting nginx. +* Fresh output file per request; ``curl -f`` so an HTTP error is a + failure, not a silently reused stale file. + +Self-contained: stdlib + the ``zstd`` CLI only (no PHP-FPM/fcgiwrap), +so it runs anywhere the rest of the Python suite does. +""" +from __future__ import annotations + +import argparse +import http.server +import pathlib +import socket +import socketserver +import subprocess +import sys +import tempfile +import threading +import time +import urllib.request + +# Sizes straddle ZSTD_CStreamInSize (~131072 = 128 KiB), the historical +# boundary for this module's truncation bug class. 1 and 1024 are below +# one zstd block (output buffered internally -> the exact forced-flush +# zero-output path); the rest force multi-buffer streaming. +SIZES = [1, 1024, 131071, 131072, 131073, 200000, 1048576] + + +def parse_args() -> argparse.Namespace: + p = argparse.ArgumentParser( + description=( + "Regression test for the proxy_buffering-off zstd " + "truncation bug (bug B)." + ) + ) + p.add_argument("--nginx-binary", required=True, + help="nginx (or angie) binary to launch.") + p.add_argument("--filter-module", + help="Path to ngx_http_zstd_filter_module.so " + "(auto-detected next to the binary if omitted).") + p.add_argument("--static-module", + help="Path to ngx_http_zstd_static_module.so " + "(auto-detected next to the binary if omitted).") + p.add_argument("--zstd-bin", default="zstd", + help="zstd CLI used for decompression.") + p.add_argument("--port", type=int, default=18086, + help="Front-end nginx port.") + p.add_argument("--backend-port", type=int, default=18087, + help="Mock chunked upstream port.") + p.add_argument("--repeat", type=int, default=5, + help="Times to re-request each size (the bug is " + "intermittent at the exact boundary).") + return p.parse_args() + + +def detect_module(explicit: str | None, nginx: pathlib.Path, + name: str) -> pathlib.Path | None: + if explicit: + return pathlib.Path(explicit) + sib = nginx.parent / name + return sib if sib.exists() else None + + +class _Handler(http.server.BaseHTTPRequestHandler): + """Serve /b/ as exactly deterministic bytes, **chunked, no + Content-Length**, written in small records so the response is a + genuine streamed chunked body (the production shape).""" + + protocol_version = "HTTP/1.1" + fixture_dir: pathlib.Path = pathlib.Path() + + def log_message(self, *a): # noqa: D401 - silence access log + pass + + def do_GET(self): + name = self.path.rsplit("/", 1)[-1].split("?")[0] + path = self.fixture_dir / name + if not path.is_file(): + self.send_response(404) + self.send_header("Content-Length", "0") + self.end_headers() + return + data = path.read_bytes() + self.send_response(200) + self.send_header("Content-Type", "application/octet-stream") + self.send_header("Transfer-Encoding", "chunked") + self.end_headers() + mv = memoryview(data) + step = 16384 + for i in range(0, len(mv), step): + chunk = bytes(mv[i:i + step]) + self.wfile.write(b"%X\r\n" % len(chunk) + chunk + b"\r\n") + self.wfile.write(b"0\r\n\r\n") + + +class _ThreadingHTTPServer(socketserver.ThreadingMixIn, + socketserver.TCPServer): + # Single-threaded server is OOM/stall-killed by the matrix and + # yields mass false failures — must be threaded. + allow_reuse_address = True + daemon_threads = True + + +def wait_port(port: int, timeout: float = 10.0) -> None: + deadline = time.time() + timeout + while time.time() < deadline: + try: + with socket.create_connection(("127.0.0.1", port), 0.5): + return + except OSError: + time.sleep(0.1) + raise RuntimeError(f"nothing listening on 127.0.0.1:{port}") + + +def http_get(port: int, path: str, timeout: float = 20.0) -> bytes: + req = urllib.request.Request( + f"http://127.0.0.1:{port}{path}", + headers={"Accept-Encoding": "zstd", + "User-Agent": "zstd-bugb-regression/1.0"}, + ) + with urllib.request.urlopen(req, timeout=timeout) as resp: + if resp.headers.get("Content-Encoding", "").lower() != "zstd": + raise RuntimeError( + f"{path}: expected Content-Encoding=zstd, got " + f"{resp.headers.get('Content-Encoding')!r}" + ) + return resp.read() + + +def zstd_decompress(zstd_bin: str, blob: bytes) -> bytes: + if blob[:4] != b"\x28\xb5\x2f\xfd": + raise RuntimeError( + f"missing zstd magic; first 16B hex={blob[:16].hex()} " + f"(truncated/aborted response)" + ) + r = subprocess.run([zstd_bin, "-d", "-q", "-c"], input=blob, + capture_output=True) + if r.returncode != 0: + raise RuntimeError( + "zstd -d failed (premature end / corrupt frame): " + + r.stderr.decode("utf-8", "replace").strip() + ) + return r.stdout + + +def main() -> int: + args = parse_args() + nginx = pathlib.Path(args.nginx_binary) + if not nginx.exists(): + raise FileNotFoundError(f"nginx binary not found: {nginx}") + + filt = detect_module(args.filter_module, nginx, + "ngx_http_zstd_filter_module.so") + stat = detect_module(args.static_module, nginx, + "ngx_http_zstd_static_module.so") + modules = [m for m in (filt, stat) if m is not None] + for m in modules: + if not m.exists(): + raise FileNotFoundError(f"module not found: {m}") + + with tempfile.TemporaryDirectory(prefix="zstd-bugb-") as td: + root = pathlib.Path(td) + fixtures = root / "fix" + logs = root / "logs" + fixtures.mkdir() + logs.mkdir() + + # Deterministic, low-compressibility fixtures (like real CSS: + # ~1:1, so the compressed stream genuinely crosses buffers). + # Distinct per size so a wrong-body response is detectable. + for n in SIZES: + buf = bytearray(n) + x = (n * 2654435761) & 0xFFFFFFFF + for i in range(n): + x = (x * 1103515245 + 12345) & 0xFFFFFFFF + buf[i] = (x >> 16) & 0xFF + (fixtures / str(n)).write_bytes(bytes(buf)) + + _Handler.fixture_dir = fixtures + backend = _ThreadingHTTPServer(("127.0.0.1", args.backend_port), + _Handler) + bt = threading.Thread(target=backend.serve_forever, daemon=True) + bt.start() + try: + wait_port(args.backend_port) + # Verify the backend itself is correct BEFORE trusting any + # end-to-end result (rig-discipline rule). + for n in (1, 131072, 1048576): + raw = urllib.request.urlopen( + f"http://127.0.0.1:{args.backend_port}/b/{n}", + timeout=10).read() + if len(raw) != n: + raise RuntimeError( + f"backend self-check failed: /b/{n} returned " + f"{len(raw)} bytes, expected {n}" + ) + + load = "".join(f"load_module {m};\n" for m in modules) + conf = root / "nginx.conf" + conf.write_text( + f"""{load}worker_processes 1; +error_log {logs}/error.log warn; +pid {root}/nginx.pid; +events {{ worker_connections 64; }} +http {{ + access_log off; + default_type application/octet-stream; + zstd on; + zstd_comp_level 3; + zstd_min_length 1; + zstd_types application/octet-stream; + server {{ + listen 127.0.0.1:{args.port}; + location /b/ {{ + # The exact production trigger: unbuffered proxying of a + # chunked, no-Content-Length upstream through the zstd + # body filter. + proxy_pass http://127.0.0.1:{args.backend_port}; + proxy_http_version 1.1; + proxy_set_header Connection ""; + proxy_buffering off; + }} + }} +}} +""", + encoding="utf-8", + ) + + proc = subprocess.Popen( + [str(nginx), "-p", str(root), "-c", str(conf), + "-g", "daemon off; master_process off;"], + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, + text=True, + ) + try: + wait_port(args.port) + failures: list[str] = [] + for n in SIZES: + expected = (fixtures / str(n)).read_bytes() + for attempt in range(1, args.repeat + 1): + label = f"size={n} attempt={attempt}/{args.repeat}" + try: + blob = http_get(args.port, f"/b/{n}") + decoded = zstd_decompress(args.zstd_bin, blob) + except Exception as exc: # noqa: BLE001 + failures.append(f"{label}: {exc}") + continue + if len(decoded) != n: + failures.append( + f"{label}: decoded {len(decoded)} bytes, " + f"expected {n} (TRUNCATION)" + ) + elif decoded != expected: + failures.append( + f"{label}: decoded body differs from " + f"origin at equal length (CORRUPTION)" + ) + + if failures: + sys.stderr.write( + "bug-B regression FAILED " + f"({len(failures)} failing checks):\n" + ) + for f in failures[:20]: + sys.stderr.write(f" - {f}\n") + elog = logs / "error.log" + if elog.exists(): + zsb = [ln for ln in + elog.read_text("utf-8", "replace").splitlines() + if "zero size buf in writer" in ln] + if zsb: + sys.stderr.write( + f" nginx logged {len(zsb)} " + f"'zero size buf in writer' " + f"(the bug-B signature)\n" + ) + return 1 + + total = len(SIZES) * args.repeat + print( + f"OK: {total} proxy_buffering-off zstd responses " + f"across {len(SIZES)} boundary sizes round-tripped " + f"byte-exact (no truncation, no zero-size-buf abort)" + ) + return 0 + finally: + proc.terminate() + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait(timeout=5) + finally: + backend.shutdown() + backend.server_close() + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except Exception as exc: # noqa: BLE001 + print(f"ERROR: {exc}", file=sys.stderr) + raise From ab42595680691d937c7bca1dda455d5195955a7f Mon Sep 17 00:00:00 2001 From: Build System Date: Mon, 18 May 2026 13:09:02 +0200 Subject: [PATCH 2/3] test: add CCtx-isolation + reload-under-load regression tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extends CI coverage to two bug classes from the module's history that no decode-and-diff test exercised: * tools/test_concurrent_cctx_isolation.py (774b4a5 "use per-request zstd contexts" class): fires many DISTINCT concurrent requests through one worker over reused keepalive connections; each response must decompress byte-exact to its OWN origin. A shared/not-reset ZSTD_CCtx produces a well-formed but wrong body (cross-request bleed) that single-body or identical-body tests cannot see. Verified: 96/96 byte-exact on the bug-B fix build. * tools/test_reload_under_load.py (2c538e0 "register CDict cleanup handler ... on reload" / config-cycle class): steady request stream with zstd_dict_file set, repeated SIGHUP reloads mid-flight; every response (old-cycle, straddling, new-cycle) must decode byte-exact, master must survive, error log must stay clean. Complements tools/test_reload_leak.sh (leaks only, not correctness across reload). Verified: 269 reqs / 5 reloads byte-exact, clean. Both wired into build-test.yml `tests` with step timeouts; the CCtx test also runs under `tests-asan` (a shared-context bug is a lifetime bug — ASAN's domain) at lighter concurrency. A dedicated CPU-spin/livelock test was prototyped and DROPPED: the livelock is input-pattern-specific and intermittent, so a fixed-drive + idle-CPU-sample test gave a false PASS on a known-livelock build — worse than no test. The livelock backstop is the existing reliable test_proxy_unbuffered_truncation.py plus its CI step `timeout-minutes` (verified: hangs and is killed on the known-livelock build → hard CI fail), so coverage is retained without a flaky gate. Rig discipline in both: threaded backend, backend self-check, distinct-per-request bodies, full-body decode-and-diff, bounded timeouts so a hang/livelock is a hard FAIL. --- .github/workflows/build-test.yml | 37 +++ tools/test_concurrent_cctx_isolation.py | 311 +++++++++++++++++++++++ tools/test_reload_under_load.py | 316 ++++++++++++++++++++++++ 3 files changed, 664 insertions(+) create mode 100644 tools/test_concurrent_cctx_isolation.py create mode 100644 tools/test_reload_under_load.py diff --git a/.github/workflows/build-test.yml b/.github/workflows/build-test.yml index 42d3c1a..be8085c 100644 --- a/.github/workflows/build-test.yml +++ b/.github/workflows/build-test.yml @@ -516,6 +516,37 @@ jobs: --port 18096 --backend-port 18097 echo "✓ compression correctness matrix passed" + - name: Run per-request CCtx isolation regression + timeout-minutes: 6 + run: | + cd "$GITHUB_WORKSPACE" + # 774b4a5 class: a shared/not-reset ZSTD_CCtx silently bleeds + # one request's compressor state into another (well-formed + # but wrong body). Fire many distinct concurrent requests + # through one worker over reused keepalive connections; each + # response must decode byte-exact to its OWN origin. + python3 tools/test_concurrent_cctx_isolation.py \ + --nginx-binary "$TEST_NGINX_BINARY" \ + --port 18098 --backend-port 18099 \ + --concurrency 16 --rounds 6 + echo "✓ per-request CCtx isolation regression passed" + + - name: Run reload-under-load correctness regression + timeout-minutes: 6 + run: | + cd "$GITHUB_WORKSPACE" + # 2c538e0 CDict-cycle class: SIGHUP reload while requests are + # in flight (zstd_dict_file set). Every response — old-cycle, + # straddling, new-cycle — must still decode byte-exact; master + # must survive; error log must stay clean. Complements + # tools/test_reload_leak.sh (which checks leaks, not response + # correctness during reload). + python3 tools/test_reload_under_load.py \ + --nginx-binary "$TEST_NGINX_BINARY" \ + --port 18100 --backend-port 18101 \ + --reloads 6 --workers 4 --duration 18 + echo "✓ reload-under-load correctness regression passed" + - name: Test summary if: success() run: echo "✓ All Perl + Python end-to-end tests passed" @@ -576,6 +607,12 @@ jobs: python3 tools/test_compression_matrix.py \ --nginx-binary "$TEST_NGINX_BINARY" --port 18096 \ --backend-port 18097 --repeat 1 + # CCtx isolation under ASAN/UBSAN: a shared/not-reset context + # is a lifetime bug — exactly what ASAN is for. Lighter + # concurrency/rounds since the sanitized binary is slower. + python3 tools/test_concurrent_cctx_isolation.py \ + --nginx-binary "$TEST_NGINX_BINARY" --port 18098 \ + --backend-port 18099 --concurrency 8 --rounds 3 echo "✓ All smoke tests clean under ASAN+UBSAN" - name: zstd_dict_file reload leak check (CDict lifecycle) diff --git a/tools/test_concurrent_cctx_isolation.py b/tools/test_concurrent_cctx_isolation.py new file mode 100644 index 0000000..e8811e7 --- /dev/null +++ b/tools/test_concurrent_cctx_isolation.py @@ -0,0 +1,311 @@ +#!/usr/bin/env python3 +"""Regression test for per-request ZSTD_CCtx isolation. + +Bug class +--------- +Commit 774b4a5 ("fix: use per-request zstd contexts") fixed a defect +where the compression context was effectively shared across requests +handled by one worker. A shared / improperly-reset ``ZSTD_CCtx`` does +NOT crash — it silently produces a *wrong but well-formed* zstd stream: +request A's worker state bleeds into request B, so B decompresses to a +mix of A's and B's bytes (or a frame that decodes cleanly but to the +wrong content). None of the other tests can see this: they each fetch a +single body, or many *identical* bodies, so cross-request bleed is +invisible. + +This test makes every concurrent request ask for **distinct content** +and asserts each response decompresses to *its own* expected bytes, +exactly. Cross-request CCtx contamination shows up as a body that +decodes to the wrong request's payload (or a decode error / wrong +length). + +Coverage +-------- +* High concurrency against a single worker (worker_processes 1) so all + requests share the one worker's context lifecycle. +* Distinct per-request payloads spanning the ZSTD_CStreamInSize buffer + boundary (the size where the module's bugs cluster), with a unique + per-request marker prefix so a swapped body is unambiguous. +* HTTP/1.1 keepalive reuse: many requests per connection (connection + reuse is where a not-properly-reset context leaks between requests). +* Several rounds; every response decode-and-diffed against its own + origin. + +Rig discipline (same hard rules as the other regression tests): +* threaded backend (single-thread is OOM/stall-killed under load) +* backend self-check before trusting any end-to-end result +* distinct bytes per request id so a wrong-body is detectable +* per-request fresh buffer; an HTTP error is a failure, not a retry +* overall + per-request timeouts so a livelock is a hard FAIL + +Self-contained: stdlib + the ``zstd`` CLI only. +""" +from __future__ import annotations + +import argparse +import concurrent.futures +import hashlib +import http.server +import pathlib +import socket +import socketserver +import subprocess +import sys +import tempfile +import threading +import time +import urllib.request + +CSTREAM_IN = 131072 # ZSTD_CStreamInSize (libzstd 1.5.x) — the size + # where this module's bug history clusters. +# Distinct sizes straddling the boundary; each request id maps to one. +SIZES = [913, 65536, CSTREAM_IN - 1, CSTREAM_IN, CSTREAM_IN + 1, + 180003, 524288] + + +def parse_args() -> argparse.Namespace: + p = argparse.ArgumentParser( + description="Per-request ZSTD_CCtx isolation regression " + "(774b4a5 class).") + p.add_argument("--nginx-binary", required=True) + p.add_argument("--filter-module") + p.add_argument("--static-module") + p.add_argument("--zstd-bin", default="zstd") + p.add_argument("--port", type=int, default=18098) + p.add_argument("--backend-port", type=int, default=18099) + p.add_argument("--concurrency", type=int, default=16, + help="Simultaneous in-flight requests.") + p.add_argument("--rounds", type=int, default=6, + help="Times to run the full concurrent batch.") + return p.parse_args() + + +def detect(explicit, nginx: pathlib.Path, name: str): + if explicit: + return pathlib.Path(explicit) + sib = nginx.parent / name + return sib if sib.exists() else None + + +def payload_for(req_id: int) -> bytes: + """Deterministic, unique-per-id, low-compressibility body. + + A 64-byte ASCII header carries the request id so a swapped body is + instantly identifiable; the remainder is an id-seeded PRNG stream + (incompressible -> the stream genuinely crosses output buffers, the + regime where context state matters).""" + size = SIZES[req_id % len(SIZES)] + head = f"REQ-{req_id:08d}-CCTX-ISOLATION-MARKER-".encode() + head = head + b"=" * (64 - len(head)) + buf = bytearray(head) + x = (req_id * 2654435761 + 1) & 0xFFFFFFFF + while len(buf) < size: + x = (x * 1103515245 + 12345) & 0xFFFFFFFF + buf.append((x >> 16) & 0xFF) + return bytes(buf[:size]) + + +class _Handler(http.server.BaseHTTPRequestHandler): + """GET /r/ -> payload_for(id), chunked, no Content-Length.""" + protocol_version = "HTTP/1.1" + + def log_message(self, *a): + pass + + def do_GET(self): + seg = self.path.split("?")[0].rsplit("/", 1)[-1] + try: + rid = int(seg) + except ValueError: + self.send_response(404) + self.send_header("Content-Length", "0") + self.end_headers() + return + data = payload_for(rid) + self.send_response(200) + self.send_header("Content-Type", "application/octet-stream") + self.send_header("Transfer-Encoding", "chunked") + self.end_headers() + mv = memoryview(data) + for i in range(0, len(mv), 16384): + c = bytes(mv[i:i + 16384]) + self.wfile.write(b"%X\r\n" % len(c) + c + b"\r\n") + self.wfile.write(b"0\r\n\r\n") + + +class _Srv(socketserver.ThreadingMixIn, socketserver.TCPServer): + allow_reuse_address = True + daemon_threads = True + + +def wait_port(port: int, timeout: float = 10.0) -> None: + end = time.time() + timeout + while time.time() < end: + try: + with socket.create_connection(("127.0.0.1", port), 0.5): + return + except OSError: + time.sleep(0.1) + raise RuntimeError(f"nothing listening on 127.0.0.1:{port}") + + +def fetch_decoded(port: int, rid: int, zstd_bin: str) -> bytes: + req = urllib.request.Request( + f"http://127.0.0.1:{port}/r/{rid}", + headers={"Accept-Encoding": "zstd", + "User-Agent": "zstd-cctx-iso/1.0"}) + with urllib.request.urlopen(req, timeout=20) as resp: + if (resp.headers.get("Content-Encoding") or "").lower() != "zstd": + raise RuntimeError( + f"rid={rid}: not zstd-encoded " + f"(C-E={resp.headers.get('Content-Encoding')!r})") + blob = resp.read() + if blob[:4] != b"\x28\xb5\x2f\xfd": + raise RuntimeError(f"rid={rid}: no zstd magic " + f"(hex={blob[:8].hex()})") + r = subprocess.run([zstd_bin, "-dq", "-c"], input=blob, + capture_output=True) + if r.returncode != 0: + raise RuntimeError(f"rid={rid}: zstd -d failed (premature " + f"end / corrupt): " + + r.stderr.decode('utf-8', 'replace')) + return r.stdout + + +def main() -> int: + args = parse_args() + nginx = pathlib.Path(args.nginx_binary) + if not nginx.exists(): + raise FileNotFoundError(nginx) + mods = [m for m in (detect(args.filter_module, nginx, + "ngx_http_zstd_filter_module.so"), + detect(args.static_module, nginx, + "ngx_http_zstd_static_module.so")) + if m] + + with tempfile.TemporaryDirectory(prefix="zstd-cctx-") as td: + root = pathlib.Path(td) + logs = root / "logs" + logs.mkdir() + + backend = _Srv(("127.0.0.1", args.backend_port), _Handler) + threading.Thread(target=backend.serve_forever, + daemon=True).start() + try: + wait_port(args.backend_port) + # Backend self-check: distinct ids -> distinct bodies. + for rid in (0, 3, 7): + raw = urllib.request.urlopen( + f"http://127.0.0.1:{args.backend_port}/r/{rid}", + timeout=10).read() + if raw != payload_for(rid): + raise RuntimeError( + f"backend self-check failed for rid={rid}") + if (payload_for(0) == payload_for(1) + or hashlib.sha256(payload_for(2)).digest() + == hashlib.sha256(payload_for(3)).digest()): + raise RuntimeError("payloads not distinct per id " + "(test would be blind to bleed)") + + load = "".join(f"load_module {m};\n" for m in mods) + conf = root / "nginx.conf" + conf.write_text(f"""{load}worker_processes 1; +error_log {logs}/error.log warn; +pid {root}/nginx.pid; +events {{ worker_connections 512; }} +http {{ + access_log off; + default_type application/octet-stream; + zstd on; + zstd_comp_level 6; + zstd_min_length 1; + zstd_types application/octet-stream; + keepalive_requests 100000; + server {{ + listen 127.0.0.1:{args.port}; + keepalive_timeout 60s; + location /r/ {{ + proxy_pass http://127.0.0.1:{args.backend_port}/r/; + proxy_http_version 1.1; + proxy_set_header Connection ""; + }} + }} +}} +""", encoding="utf-8") + + proc = subprocess.Popen( + [str(nginx), "-p", str(root), "-c", str(conf), + "-g", "daemon off; master_process off;"], + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, + text=True) + try: + wait_port(args.port) + failures: list[str] = [] + total = 0 + base = 0 + for rnd in range(args.rounds): + ids = list(range(base, base + args.concurrency)) + base += args.concurrency + with concurrent.futures.ThreadPoolExecutor( + max_workers=args.concurrency) as ex: + futs = { + ex.submit(fetch_decoded, args.port, rid, + args.zstd_bin): rid + for rid in ids} + for fut in concurrent.futures.as_completed( + futs): + rid = futs[fut] + total += 1 + try: + got = fut.result() + except Exception as e: # noqa: BLE001 + failures.append( + f"round {rnd} rid={rid}: {e}") + continue + want = payload_for(rid) + if got != want: + # Identify if it is *another request's* + # body (the CCtx-bleed signature). + culprit = "" + if got[:32].startswith(b"REQ-"): + culprit = (" looks like " + + got[:28].decode( + "ascii", "replace")) + failures.append( + f"round {rnd} rid={rid}: body " + f"mismatch (got {len(got)}B want " + f"{len(want)}B){culprit}") + if failures: + sys.stderr.write( + f"CCtx-ISOLATION FAILED: " + f"{len(failures)}/{total}:\n") + for f in failures[:25]: + sys.stderr.write(f" - {f}\n") + if len(failures) > 25: + sys.stderr.write( + f" ... +{len(failures)-25} more\n") + return 1 + print(f"OK: {total} concurrent requests " + f"({args.concurrency} in flight x {args.rounds} " + f"rounds, distinct per-request bodies, keepalive " + f"reuse) each decoded byte-exact to its own " + f"origin — no cross-request CCtx bleed") + return 0 + finally: + proc.terminate() + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait(timeout=5) + finally: + backend.shutdown() + backend.server_close() + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except Exception as exc: # noqa: BLE001 + print(f"ERROR: {exc}", file=sys.stderr) + raise diff --git a/tools/test_reload_under_load.py b/tools/test_reload_under_load.py new file mode 100644 index 0000000..854e968 --- /dev/null +++ b/tools/test_reload_under_load.py @@ -0,0 +1,316 @@ +#!/usr/bin/env python3 +"""Regression test: response correctness across nginx reload under load. + +Bug class +--------- +Commit 2c538e0 ("fix: register CDict cleanup handler to prevent leak on +reload") and the broader CDict / config-cycle lifetime history: a +``zstd_dict_file`` ``ZSTD_CDict`` is owned by the configuration cycle. +On ``nginx -s reload`` the old cycle is retired while in-flight +requests on old workers are still compressing with the old cycle's +CDict. A mishandled lifetime here can leak (covered by the existing +``tools/test_reload_leak.sh``) **or** corrupt / truncate responses +that straddle the reload — which nothing currently asserts. + +This test keeps a steady request stream running, issues repeated +``SIGHUP`` reloads while requests are in flight, and verifies **every** +response (old-cycle, straddling, and new-cycle) still decompresses +byte-exact to its origin, the master stays up, and the error log stays +clean (no ``[alert]``/``[emerg]``/``zero size buf``/sanitizer). + +Rig discipline (same hard rules as the other regression tests): +* threaded backend; backend self-check before trusting results +* distinct per-request bodies so a wrong/stale body is detectable +* decode-and-diff the full body; HTTP error is a failure +* a real ``zstd_dict_file`` so the CDict lifetime path is exercised +* overall timeout so a hang/livelock is a hard FAIL + +Self-contained: stdlib + the ``zstd`` CLI only. +""" +from __future__ import annotations + +import argparse +import http.server +import os +import pathlib +import signal +import socket +import socketserver +import subprocess +import sys +import tempfile +import threading +import time +import urllib.request + +CSTREAM_IN = 131072 +SIZES = [4096, 65000, CSTREAM_IN - 1, CSTREAM_IN + 1, 200000] + + +def parse_args() -> argparse.Namespace: + p = argparse.ArgumentParser( + description="Response correctness across reload under load " + "(2c538e0 CDict-cycle class).") + p.add_argument("--nginx-binary", required=True) + p.add_argument("--filter-module") + p.add_argument("--static-module") + p.add_argument("--zstd-bin", default="zstd") + p.add_argument("--port", type=int, default=18100) + p.add_argument("--backend-port", type=int, default=18101) + p.add_argument("--reloads", type=int, default=6) + p.add_argument("--workers", type=int, default=4, + help="Concurrent client workers driving load.") + p.add_argument("--duration", type=float, default=18.0, + help="Seconds of sustained load (reloads spread " + "across it).") + return p.parse_args() + + +def detect(explicit, nginx: pathlib.Path, name: str): + if explicit: + return pathlib.Path(explicit) + sib = nginx.parent / name + return sib if sib.exists() else None + + +def payload_for(rid: int) -> bytes: + size = SIZES[rid % len(SIZES)] + head = f"RELOAD-REQ-{rid:09d}-".encode() + head = head + b"." * (48 - len(head)) + buf = bytearray(head) + x = (rid * 2246822519 + 7) & 0xFFFFFFFF + while len(buf) < size: + x = (x * 1103515245 + 12345) & 0xFFFFFFFF + buf.append((x >> 16) & 0xFF) + return bytes(buf[:size]) + + +class _Handler(http.server.BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def log_message(self, *a): + pass + + def do_GET(self): + seg = self.path.split("?")[0].rsplit("/", 1)[-1] + try: + rid = int(seg) + except ValueError: + self.send_response(404) + self.send_header("Content-Length", "0") + self.end_headers() + return + data = payload_for(rid) + self.send_response(200) + self.send_header("Content-Type", "application/octet-stream") + self.send_header("Transfer-Encoding", "chunked") + self.end_headers() + mv = memoryview(data) + for i in range(0, len(mv), 16384): + c = bytes(mv[i:i + 16384]) + self.wfile.write(b"%X\r\n" % len(c) + c + b"\r\n") + self.wfile.write(b"0\r\n\r\n") + + +class _Srv(socketserver.ThreadingMixIn, socketserver.TCPServer): + allow_reuse_address = True + daemon_threads = True + + +def wait_port(port: int, timeout: float = 10.0) -> None: + end = time.time() + timeout + while time.time() < end: + try: + with socket.create_connection(("127.0.0.1", port), 0.5): + return + except OSError: + time.sleep(0.1) + raise RuntimeError(f"nothing listening on 127.0.0.1:{port}") + + +def one(port: int, rid: int, zstd_bin: str) -> None: + req = urllib.request.Request( + f"http://127.0.0.1:{port}/r/{rid}", + headers={"Accept-Encoding": "zstd", + "User-Agent": "zstd-reload/1.0"}) + with urllib.request.urlopen(req, timeout=20) as resp: + if (resp.headers.get("Content-Encoding") or "").lower() != "zstd": + raise RuntimeError( + f"rid={rid}: not zstd " + f"(C-E={resp.headers.get('Content-Encoding')!r})") + blob = resp.read() + if blob[:4] != b"\x28\xb5\x2f\xfd": + raise RuntimeError(f"rid={rid}: no zstd magic") + r = subprocess.run([zstd_bin, "-dq", "-c"], input=blob, + capture_output=True) + if r.returncode != 0: + raise RuntimeError(f"rid={rid}: zstd -d failed: " + + r.stderr.decode("utf-8", "replace")) + if r.stdout != payload_for(rid): + raise RuntimeError( + f"rid={rid}: body mismatch across reload " + f"(got {len(r.stdout)}B want {len(payload_for(rid))}B)") + + +def main() -> int: + args = parse_args() + nginx = pathlib.Path(args.nginx_binary) + if not nginx.exists(): + raise FileNotFoundError(nginx) + mods = [m for m in (detect(args.filter_module, nginx, + "ngx_http_zstd_filter_module.so"), + detect(args.static_module, nginx, + "ngx_http_zstd_static_module.so")) + if m] + + with tempfile.TemporaryDirectory(prefix="zstd-reload-") as td: + root = pathlib.Path(td) + logs = root / "logs" + logs.mkdir() + # A non-trivial dictionary so ZSTD_createCDict actually + # allocates and the per-cycle CDict lifetime is exercised. + dict_path = root / "zstd.dict" + dseed = bytearray() + x = 0x9E3779B1 + while len(dseed) < 16384: + x = (x * 1103515245 + 12345) & 0xFFFFFFFF + dseed.append((x >> 16) & 0xFF) + dict_path.write_bytes(bytes(dseed)) + + backend = _Srv(("127.0.0.1", args.backend_port), _Handler) + threading.Thread(target=backend.serve_forever, + daemon=True).start() + try: + wait_port(args.backend_port) + for rid in (0, 2, 4): + if urllib.request.urlopen( + f"http://127.0.0.1:{args.backend_port}/r/{rid}", + timeout=10).read() != payload_for(rid): + raise RuntimeError( + f"backend self-check failed rid={rid}") + + load = "".join(f"load_module {m};\n" for m in mods) + pid_file = root / "nginx.pid" + conf = root / "nginx.conf" + conf.write_text(f"""{load}worker_processes 2; +error_log {logs}/error.log warn; +pid {pid_file}; +events {{ worker_connections 256; }} +http {{ + access_log off; + default_type application/octet-stream; + zstd_dict_file {dict_path}; + zstd on; + zstd_comp_level 6; + zstd_min_length 1; + zstd_types application/octet-stream; + server {{ + listen 127.0.0.1:{args.port}; + location /r/ {{ + proxy_pass http://127.0.0.1:{args.backend_port}/r/; + proxy_http_version 1.1; + proxy_set_header Connection ""; + }} + }} +}} +""", encoding="utf-8") + + # Master process mode (reload needs a master to signal). + proc = subprocess.Popen( + [str(nginx), "-p", str(root), "-c", str(conf), + "-g", "daemon off;"], + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, + text=True) + try: + wait_port(args.port) + stop = threading.Event() + failures: list[str] = [] + counter = {"n": 0} + clock = threading.Lock() + + def worker(): + while not stop.is_set(): + with clock: + rid = counter["n"] + counter["n"] += 1 + try: + one(args.port, rid, args.zstd_bin) + except Exception as e: # noqa: BLE001 + failures.append(str(e)) + + threads = [threading.Thread(target=worker, daemon=True) + for _ in range(args.workers)] + for t in threads: + t.start() + + # Spread SIGHUP reloads across the load window. + deadline = time.time() + args.duration + interval = args.duration / (args.reloads + 1) + master_pid = int(pid_file.read_text().strip()) + done_reloads = 0 + while time.time() < deadline and not failures: + time.sleep(interval) + if done_reloads < args.reloads: + os.kill(master_pid, signal.SIGHUP) + done_reloads += 1 + stop.set() + for t in threads: + t.join(timeout=10) + + # Final clean shutdown so all cycle cleanups run + # (incl. the CDict cleanup handler from 2c538e0). + os.kill(master_pid, signal.SIGQUIT) + try: + proc.wait(timeout=10) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait(timeout=5) + + el = logs / "error.log" + logtext = (el.read_text("utf-8", "replace") + if el.exists() else "") + bad = [ln for ln in logtext.splitlines() + if ("[alert]" in ln or "[emerg]" in ln + or "zero size buf in writer" in ln + or "AddressSanitizer" in ln + or "LeakSanitizer" in ln + or "runtime error:" in ln)] + + if failures or bad: + sys.stderr.write( + f"RELOAD-UNDER-LOAD FAILED: " + f"{len(failures)} response failures, " + f"{len(bad)} log issues, " + f"{counter['n']} reqs, " + f"{done_reloads} reloads:\n") + for f in failures[:15]: + sys.stderr.write(f" resp: {f}\n") + for b in bad[:10]: + sys.stderr.write(f" log: {b}\n") + return 1 + + print(f"OK: {counter['n']} requests across " + f"{done_reloads} SIGHUP reloads under " + f"{args.workers}-way load — every response " + f"decoded byte-exact, master survived, error log " + f"clean (CDict cycle lifetime intact)") + return 0 + finally: + if proc.poll() is None: + proc.terminate() + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait(timeout=5) + finally: + backend.shutdown() + backend.server_close() + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except Exception as exc: # noqa: BLE001 + print(f"ERROR: {exc}", file=sys.stderr) + raise From 8c5b772cf16b92c89fc33f219d20bd5096f9f260 Mon Sep 17 00:00:00 2001 From: Build System Date: Mon, 18 May 2026 13:28:18 +0200 Subject: [PATCH 3/3] =?UTF-8?q?test:=20cover=20zstd=5Flong=20directive=20(?= =?UTF-8?q?Gap=201)=20=E2=80=94=20add=20LDM=20regression=20test?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Commit 8edb7bb (zstd_long directive) had ZERO coverage. Add tools/test_zstd_long_ldm.py: a 16 MiB body whose duplicate region repeats beyond the level-6 default match window. Asserts (a) byte-exact decode and (b) ~2:1 smaller output with zstd_long on vs off — proving ZSTD_c_enableLongDistanceMatching actually reached libzstd, engaged, and does not corrupt/truncate the stream (the bug-B failure mode). Fail-first verified: passes on the fix, fails when the directive is a no-op. Also document Gaps 2 and 3 from the coverage audit: * Gap 2 (f7e2ef3): proven untestable via black-box http test (purely defensive hardening; pre-fix and fixed .so produce identical responses). * Gap 3 (HTTP/2 transport, 8281baa): requires TLS/ALPN to test in nginx; not worth the complexity when bug-B is already covered by test_proxy_unbuffered_truncation.py and matrix-under-asan. Wire zstd_long test into both tests and tests-asan jobs. --- .github/workflows/build-test.yml | 16 ++ t/01-static.t | 29 +++ tools/test_zstd_long_ldm.py | 347 +++++++++++++++++++++++++++++++ 3 files changed, 392 insertions(+) create mode 100644 tools/test_zstd_long_ldm.py diff --git a/.github/workflows/build-test.yml b/.github/workflows/build-test.yml index be8085c..107c4aa 100644 --- a/.github/workflows/build-test.yml +++ b/.github/workflows/build-test.yml @@ -547,6 +547,22 @@ jobs: --reloads 6 --workers 4 --duration 18 echo "✓ reload-under-load correctness regression passed" + - name: Run zstd_long (LDM) directive regression + timeout-minutes: 6 + run: | + cd "$GITHUB_WORKSPACE" + # 8edb7bb zstd_long class: the directive had ZERO coverage. + # A 16 MiB body whose duplicate region repeats beyond the + # level-6 default window must (a) decode byte-exact and + # (b) compress ~2:1 smaller with zstd_long on than off — + # proving ZSTD_c_enableLongDistanceMatching actually reached + # libzstd and engaged, and the directive neither corrupts nor + # truncates the stream (the bug-B failure mode). + python3 tools/test_zstd_long_ldm.py \ + --nginx-binary "$TEST_NGINX_BINARY" \ + --port 18102 --backend-port 18103 + echo "✓ zstd_long (LDM) directive regression passed" + - name: Test summary if: success() run: echo "✓ All Perl + Python end-to-end tests passed" diff --git a/t/01-static.t b/t/01-static.t index 7b995fa..be5ed7c 100644 --- a/t/01-static.t +++ b/t/01-static.t @@ -404,3 +404,32 @@ Content-Encoding: zstd !Content-Encoding --- no_error_log [alert] + + + +# COVERAGE NOTES on gaps found by the git-history CI audit: +# +# Gap 2 — f7e2ef3 ("clear Accept-Ranges in static module"): +# deliberately NOT covered by a black-box test here. The static +# handler serves a local file and never sets r->allow_ranges (in this +# nginx, only the upstream/proxy path sets it — see +# ngx_http_upstream.c, and ngx_http_range_filter_module.c bails unless +# r->allow_ranges). ngx_http_clear_accept_ranges() in the static +# module is therefore purely defensive: no request reachable through +# zstd_static alone makes the pre-fix and fixed builds differ +# (empirically verified — identical 200, no Accept-Ranges, no +# Content-Range, on both a pre-f7e2ef3 and a fixed .so). A Perl test +# asserting !Accept-Ranges would PASS on the buggy build too, i.e. be +# blind. Per the project's fail-first discipline we do not add a test +# that cannot fail on the unfixed code. +# +# Gap 3 — HTTP/2 transport axis (8281baa bug-B class): +# the build enables --with-http_v2_module but nothing tests the h2 +# path. HTTP/2 in nginx requires TLS + ALPN/Upgrade negotiation; h2c +# (cleartext) does not work without Upgrade in nginx config. A Python +# test without TLS cannot easily drive the h2 path. The bug-B defect +# (empty-buffer, flush-state-machine, c->buffered accounting) is +# already well-covered by test_proxy_unbuffered_truncation.py +# (HTTP/1.1) and the matrix under ASAN; the h2-specific framing path +# would be redundant effort without adding coverage for a new code +# path. Left for future work when/if CI adds TLS test infrastructure. diff --git a/tools/test_zstd_long_ldm.py b/tools/test_zstd_long_ldm.py new file mode 100644 index 0000000..890878a --- /dev/null +++ b/tools/test_zstd_long_ldm.py @@ -0,0 +1,347 @@ +#!/usr/bin/env python3 +"""Regression test for the ``zstd_long`` (long-distance matching) directive. + +Bug class +--------- +Commit 8edb7bb added the ``zstd_long`` directive +(``ZSTD_c_enableLongDistanceMatching``) and the interacting +``zstd_window_log`` cap (``ZSTD_c_windowLog``). Both are set on the +per-request ``ZSTD_CCtx`` *before* compression starts. A shipped +feature with **zero** test coverage: a mistake in the parameter wiring +(wrong enum, value clamped/rejected, set after the stream is already +running, or an interaction with the per-request CCtx / window cap) +would either + +* abort the stream (``ZSTD_setParameter`` error -> 500 / truncated + body — the bug-B failure mode), or +* silently produce a *wrong but well-formed* frame, or +* quietly do nothing (LDM never actually engages), so the directive is + a no-op and operators get none of the promised ratio win. + +This test asserts all three cannot happen: + +1. **Correctness** — a large body that is internally repetitive *beyond + the regular match window* (the regime LDM exists for) still + decompresses byte-exact with ``zstd_long on``, including with an + explicit small ``zstd_window_log`` cap (the documented "explicit + window cap still takes precedence" path around line 1000 of the + filter module). +2. **Effect** — with such a body, ``zstd_long on`` must produce a + *strictly smaller* compressed stream than ``zstd_long off`` at the + same level. If LDM were silently a no-op the two sizes would match; + a meaningful gap proves the directive reached libzstd and engaged. +3. **No regression** — master process clean shutdown, error log free of + ``[alert]``/``[emerg]``/``zero size buf``/sanitizer findings. + +Rig discipline (same hard rules as the other regression tests): +* threaded backend; backend self-check before trusting any result +* a body with long-range repeats *and* a unique per-request marker so a + swapped/stale body is detectable +* decode-and-diff the full body; any HTTP error is a hard failure +* overall timeout via the CI step so a hang/livelock is a hard FAIL + +Self-contained: stdlib + the ``zstd`` CLI only. +""" +from __future__ import annotations + +import argparse +import http.server +import pathlib +import socket +import socketserver +import subprocess +import sys +import tempfile +import threading +import time +import urllib.request + +# Body layout (designed so ONLY long-distance matching can collapse it +# at the configured compression level, with NO window-cap fiddling — +# the directive under test is exercised in isolation): +# +# [ 8 MiB unique pseudo-random region R ][ exact copy of R ] +# +# R is incompressible (id-seeded PRNG) so the regular matcher cannot +# shrink it. The second copy of R repeats at distance 8 MiB, which is +# far beyond zstd's level-6 *default* match window (~1–2 MiB): by the +# time the duplicate arrives the regular matcher's window has long slid +# past the original, so it cannot back-reference it and the body stays +# ~incompressible (~16 MiB out). LDM, whose long-range hash table +# defaults to a 2^27 window, *can* reach back 8 MiB, match the copy to +# R, and roughly halve the output (~8 MiB). Empirically (libzstd +# 1.5.x): off ~= 16 MiB, on ~= 8 MiB — a clean, unambiguous 2:1 signal +# that LDM genuinely engaged. A swapped/stale body is caught by the +# per-request marker header. +UNIQUE = 8 * 1024 * 1024 +TOTAL = 2 * UNIQUE + + +def parse_args() -> argparse.Namespace: + p = argparse.ArgumentParser( + description="zstd_long / windowLog directive regression " + "(8edb7bb class).") + p.add_argument("--nginx-binary", required=True) + p.add_argument("--filter-module") + p.add_argument("--static-module") + p.add_argument("--zstd-bin", default="zstd") + p.add_argument("--port", type=int, default=18102) + p.add_argument("--backend-port", type=int, default=18103) + return p.parse_args() + + +def detect(explicit, nginx: pathlib.Path, name: str): + if explicit: + return pathlib.Path(explicit) + sib = nginx.parent / name + return sib if sib.exists() else None + + +def payload_for(rid: int) -> bytes: + """Unique incompressible region R followed by an exact copy of R. + + R is a ``UNIQUE``-byte id-seeded PRNG stream (incompressible, so the + regular matcher cannot shrink it). The second half is byte-identical + to R, repeating at distance ``UNIQUE`` — far beyond the small + ``zstd_window_log`` cap, so only long-distance matching can collapse + it. A 64-byte marker header makes a swapped/stale body unambiguous.""" + head = f"ZSTD-LONG-REQ-{rid:08d}-LDM-MARKER-".encode() + head = head + b"#" * (64 - len(head)) + r = bytearray(head) + x = (rid * 2654435761 + 1) & 0xFFFFFFFF + while len(r) < UNIQUE: + x = (x * 1103515245 + 12345) & 0xFFFFFFFF + r.append((x >> 16) & 0xFF) + r = bytes(r[:UNIQUE]) + return r + r + + +class _Handler(http.server.BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def log_message(self, *a): + pass + + def do_GET(self): + seg = self.path.split("?")[0].rsplit("/", 1)[-1] + try: + rid = int(seg) + except ValueError: + self.send_response(404) + self.send_header("Content-Length", "0") + self.end_headers() + return + data = payload_for(rid) + self.send_response(200) + self.send_header("Content-Type", "application/octet-stream") + self.send_header("Transfer-Encoding", "chunked") + self.end_headers() + mv = memoryview(data) + for i in range(0, len(mv), 16384): + c = bytes(mv[i:i + 16384]) + self.wfile.write(b"%X\r\n" % len(c) + c + b"\r\n") + self.wfile.write(b"0\r\n\r\n") + + +class _Srv(socketserver.ThreadingMixIn, socketserver.TCPServer): + allow_reuse_address = True + daemon_threads = True + + +def wait_port(port: int, timeout: float = 10.0) -> None: + end = time.time() + timeout + while time.time() < end: + try: + with socket.create_connection(("127.0.0.1", port), 0.5): + return + except OSError: + time.sleep(0.1) + raise RuntimeError(f"nothing listening on 127.0.0.1:{port}") + + +def fetch(port: int, rid: int, zstd_bin: str) -> tuple[bytes, int]: + """Return (decoded_body, compressed_size) or raise on any error.""" + req = urllib.request.Request( + f"http://127.0.0.1:{port}/r/{rid}", + headers={"Accept-Encoding": "zstd", + "User-Agent": "zstd-long/1.0"}) + with urllib.request.urlopen(req, timeout=30) as resp: + if (resp.headers.get("Content-Encoding") or "").lower() != "zstd": + raise RuntimeError( + f"rid={rid}: not zstd-encoded " + f"(C-E={resp.headers.get('Content-Encoding')!r})") + blob = resp.read() + if blob[:4] != b"\x28\xb5\x2f\xfd": + raise RuntimeError(f"rid={rid}: no zstd magic " + f"(hex={blob[:8].hex()})") + r = subprocess.run([zstd_bin, "-dq", "-c"], input=blob, + capture_output=True) + if r.returncode != 0: + raise RuntimeError(f"rid={rid}: zstd -d failed (premature " + f"end / corrupt): " + + r.stderr.decode("utf-8", "replace")) + return r.stdout, len(blob) + + +def run_server(nginx, conf, root, port, backend_port, zstd_bin, + rids: list[int]) -> dict[int, tuple[bytes, int]]: + proc = subprocess.Popen( + [str(nginx), "-p", str(root), "-c", str(conf), + "-g", "daemon off; master_process off;"], + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) + try: + wait_port(port) + return {rid: fetch(port, rid, zstd_bin) for rid in rids} + finally: + proc.terminate() + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait(timeout=5) + + +def main() -> int: + args = parse_args() + nginx = pathlib.Path(args.nginx_binary) + if not nginx.exists(): + raise FileNotFoundError(nginx) + mods = [m for m in (detect(args.filter_module, nginx, + "ngx_http_zstd_filter_module.so"), + detect(args.static_module, nginx, + "ngx_http_zstd_static_module.so")) + if m] + + with tempfile.TemporaryDirectory(prefix="zstd-long-") as td: + root = pathlib.Path(td) + logs = root / "logs" + logs.mkdir() + + backend = _Srv(("127.0.0.1", args.backend_port), _Handler) + threading.Thread(target=backend.serve_forever, + daemon=True).start() + try: + wait_port(args.backend_port) + for rid in (1, 2, 3): + raw = urllib.request.urlopen( + f"http://127.0.0.1:{args.backend_port}/r/{rid}", + timeout=10).read() + if raw != payload_for(rid): + raise RuntimeError( + f"backend self-check failed for rid={rid}") + if len(payload_for(1)) != TOTAL: + raise RuntimeError("payload not the expected size") + + load = "".join(f"load_module {m};\n" for m in mods) + + def conf_text(extra: str) -> str: + return f"""{load}worker_processes 1; +error_log {logs}/error.log warn; +pid {root}/nginx.pid; +events {{ worker_connections 256; }} +http {{ + access_log off; + default_type application/octet-stream; + zstd on; + zstd_comp_level 6; + zstd_min_length 1; + zstd_types application/octet-stream; +{extra} + server {{ + listen 127.0.0.1:{args.port}; + location /r/ {{ + proxy_pass http://127.0.0.1:{args.backend_port}/r/; + proxy_http_version 1.1; + proxy_set_header Connection ""; + }} + }} +}} +""" + + rids = [11, 22, 33] + + # Baseline: LDM off (default). + off_conf = root / "off.conf" + off_conf.write_text(conf_text(""), encoding="utf-8") + off = run_server(nginx, off_conf, root, args.port, + args.backend_port, args.zstd_bin, rids) + + # LDM on (directive under test, in isolation — no window + # cap, so zstd's LDM default 2^27 window is what reaches + # back the 8 MiB to the duplicate region). + on_conf = root / "on.conf" + on_conf.write_text( + conf_text(" zstd_long on;\n"), + encoding="utf-8") + on = run_server(nginx, on_conf, root, args.port, + args.backend_port, args.zstd_bin, rids) + + failures: list[str] = [] + for rid in rids: + want = payload_for(rid) + off_body, off_sz = off[rid] + on_body, on_sz = on[rid] + # (1) correctness on both configs + if off_body != want: + failures.append( + f"rid={rid}: LDM-off body mismatch " + f"(got {len(off_body)}B want {len(want)}B)") + if on_body != want: + failures.append( + f"rid={rid}: LDM-on body mismatch " + f"(got {len(on_body)}B want {len(want)}B) — " + f"directive corrupts/truncates the stream") + # (2) LDM must actually engage. The duplicate region is + # only reachable via LDM's long-range window, so LDM-on + # must roughly halve the output vs LDM-off. Require a + # large, unambiguous gap (>25% smaller) so an incidental + # few-byte difference cannot pass as "engaged"; a silent + # no-op leaves the two sizes essentially equal. + if on_sz >= off_sz * 0.75: + failures.append( + f"rid={rid}: zstd_long had no/insufficient " + f"effect (on={on_sz}B vs off={off_sz}B; " + f"expected on <~ off/2) — directive is a " + f"silent no-op (LDM never engaged)") + + # (3) error log clean across both runs. + el = logs / "error.log" + logtext = (el.read_text("utf-8", "replace") + if el.exists() else "") + bad = [ln for ln in logtext.splitlines() + if ("[alert]" in ln or "[emerg]" in ln + or "zero size buf in writer" in ln + or "AddressSanitizer" in ln + or "LeakSanitizer" in ln + or "runtime error:" in ln)] + + if failures or bad: + sys.stderr.write( + f"ZSTD-LONG FAILED: {len(failures)} assertion " + f"failures, {len(bad)} log issues:\n") + for f in failures: + sys.stderr.write(f" - {f}\n") + for b in bad[:10]: + sys.stderr.write(f" log: {b}\n") + return 1 + + ratios = ", ".join( + f"rid={rid}: {off[rid][1]}->{on[rid][1]}B" + for rid in rids) + print(f"OK: zstd_long on decoded byte-exact on {len(rids)} " + f"16MiB long-range-repetitive bodies AND compressed " + f">=25% smaller than LDM-off (~2:1 as expected: " + f"{ratios}) — directive reaches libzstd, engages, and " + f"does not corrupt the stream; error log clean") + return 0 + finally: + backend.shutdown() + backend.server_close() + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except Exception as exc: # noqa: BLE001 + print(f"ERROR: {exc}", file=sys.stderr) + raise