mirror of
https://github.com/hanzoai/datastore.git
synced 2026-09-26 13:55:07 +00:00
Bound the zstd CLI decompression fallback with a timeout
Both fallback paths could hang the whole report on a wedged zstd: - fetch_perf_report.py: proc.stderr.read() blocks until the child exits, so proc.wait(timeout=120) was never reached. Replace it with a threading.Timer watchdog that kills the child, which unblocks both the stdin write and the stderr read and turns a stuck shard into the normal per-shard error. Add a fast regression test using a fake, wedged zstd on PATH and a shrunk timeout. - fetch_ci_report.js: add a 120s timeout to the execFileSync zstd fallback so a wedged child can no longer block the helper indefinitely. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
co-authored by
Claude Opus 4.8
parent
ce2e943be5
commit
4fe6336cb6
@@ -57,7 +57,8 @@ function maybeDecompress(buf) {
|
||||
if (buf.length >= 4 && buf[0] === 0x28 && buf[1] === 0xb5 && buf[2] === 0x2f && buf[3] === 0xfd) {
|
||||
if (typeof zlib.zstdDecompressSync === 'function') return zlib.zstdDecompressSync(buf);
|
||||
const { execFileSync } = require('child_process');
|
||||
return execFileSync('zstd', ['-dcq'], { input: buf, maxBuffer: 2 * 1024 * 1024 * 1024 });
|
||||
// Bound the child: a wedged zstd would otherwise block the whole helper indefinitely.
|
||||
return execFileSync('zstd', ['-dcq'], { input: buf, maxBuffer: 2 * 1024 * 1024 * 1024, timeout: 120000 });
|
||||
}
|
||||
if (buf.length >= 2 && buf[0] === 0x1f && buf[1] === 0x8b) {
|
||||
return zlib.gunzipSync(buf);
|
||||
|
||||
@@ -43,6 +43,7 @@ import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
from urllib.parse import urlparse, parse_qs
|
||||
from urllib.request import urlopen
|
||||
@@ -116,6 +117,9 @@ class _PrefixedReader:
|
||||
return data
|
||||
|
||||
|
||||
_ZSTD_CLI_TIMEOUT_SEC = 120 # watchdog bound for the zstd-CLI decompression fallback
|
||||
|
||||
|
||||
def _stream_to_file(resp, dest):
|
||||
"""Stream `resp` to `dest`, transparently decompressing zstd/gzip by magic bytes.
|
||||
|
||||
@@ -132,17 +136,27 @@ def _stream_to_file(resp, dest):
|
||||
except ImportError:
|
||||
proc = subprocess.Popen(["zstd", "-dcq"], stdin=subprocess.PIPE, stdout=f,
|
||||
stderr=subprocess.PIPE)
|
||||
# Feed the body in and wait for the child directly: communicate() would try to
|
||||
# flush the already-closed stdin, which raises ValueError on most Python versions.
|
||||
# A watchdog bounds the whole exchange: both the stdin write and the stderr read
|
||||
# block indefinitely if the child wedges, so wait(timeout=) alone never fires.
|
||||
# Killing the child unblocks both and turns a stuck shard into a normal error.
|
||||
# (communicate() is unusable here: it flushes the already-closed stdin, which
|
||||
# raises ValueError on most Python versions.)
|
||||
watchdog = threading.Timer(_ZSTD_CLI_TIMEOUT_SEC, proc.kill)
|
||||
watchdog.start()
|
||||
try:
|
||||
shutil.copyfileobj(source, proc.stdin)
|
||||
proc.stdin.close()
|
||||
except BrokenPipeError:
|
||||
pass # zstd exited early (e.g. corrupt input); the real error is on stderr
|
||||
stderr = proc.stderr.read()
|
||||
proc.stderr.close()
|
||||
if proc.wait(timeout=120) != 0:
|
||||
raise RuntimeError(f"zstd decompression failed: {stderr.decode('utf-8', 'replace')}")
|
||||
try:
|
||||
shutil.copyfileobj(source, proc.stdin)
|
||||
proc.stdin.close()
|
||||
except BrokenPipeError:
|
||||
pass # zstd exited early (corrupt input or killed); real error is on stderr
|
||||
stderr = proc.stderr.read()
|
||||
proc.stderr.close()
|
||||
rc = proc.wait()
|
||||
finally:
|
||||
watchdog.cancel()
|
||||
if rc != 0:
|
||||
raise RuntimeError(f"zstd decompression failed (exit {rc}): "
|
||||
f"{stderr.decode('utf-8', 'replace')}")
|
||||
elif header[:2] == b"\x1f\x8b": # gzip magic
|
||||
with gzip.GzipFile(fileobj=source) as gz:
|
||||
shutil.copyfileobj(gz, f)
|
||||
|
||||
@@ -207,6 +207,45 @@ def test_stream_to_file_zstd_cli_fallback():
|
||||
_sys.modules["zstandard"] = saved
|
||||
|
||||
|
||||
def test_stream_to_file_zstd_cli_times_out():
|
||||
# A wedged zstd must not hang the report forever: the watchdog kills it and the failure
|
||||
# becomes a normal per-shard error. Uses a fake `zstd` on PATH that ignores stdin and never
|
||||
# exits, plus a shrunk timeout so the test stays fast.
|
||||
import io as _io
|
||||
import stat as _stat
|
||||
import sys as _sys
|
||||
import time as _time
|
||||
|
||||
saved_mod = _sys.modules.get("zstandard", "MISSING")
|
||||
saved_path = os.environ["PATH"]
|
||||
saved_timeout = fpr._ZSTD_CLI_TIMEOUT_SEC
|
||||
_sys.modules["zstandard"] = None # force the CLI fallback
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
fake = os.path.join(tmp, "zstd")
|
||||
with open(fake, "w") as f:
|
||||
f.write("#!/bin/sh\nexec sleep 300\n")
|
||||
os.chmod(fake, os.stat(fake).st_mode | _stat.S_IEXEC)
|
||||
os.environ["PATH"] = tmp + os.pathsep + saved_path
|
||||
fpr._ZSTD_CLI_TIMEOUT_SEC = 2
|
||||
try:
|
||||
zst_magic_input = b"\x28\xb5\x2f\xfd" + b"\x00" * 4096
|
||||
dest = os.path.join(tmp, "out")
|
||||
t0 = _time.time()
|
||||
try:
|
||||
fpr._stream_to_file(_io.BytesIO(zst_magic_input), dest)
|
||||
assert False, "expected the wedged zstd to raise"
|
||||
except RuntimeError as e:
|
||||
assert "zstd decompression failed" in str(e), e
|
||||
assert _time.time() - t0 < 30, "watchdog did not bound the wedged child"
|
||||
finally:
|
||||
fpr._ZSTD_CLI_TIMEOUT_SEC = saved_timeout
|
||||
os.environ["PATH"] = saved_path
|
||||
if saved_mod == "MISSING":
|
||||
_sys.modules.pop("zstandard", None)
|
||||
else:
|
||||
_sys.modules["zstandard"] = saved_mod
|
||||
|
||||
|
||||
def test_prefixed_reader_reassembles_stream():
|
||||
import io as _io
|
||||
|
||||
@@ -248,6 +287,7 @@ if __name__ == "__main__":
|
||||
test_maybe_decompress_handles_plain_gzip_zstd()
|
||||
test_stream_to_file_handles_plain_gzip_zstd()
|
||||
test_stream_to_file_zstd_cli_fallback()
|
||||
test_stream_to_file_zstd_cli_times_out()
|
||||
test_prefixed_reader_reassembles_stream()
|
||||
test_download_shard_isolates_failures()
|
||||
print("All fetch_perf_report tests passed (or skipped).")
|
||||
|
||||
Reference in New Issue
Block a user