From c9d31bdf5603537121fac9b127e4b70e94ff66d3 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Mon, 19 Oct 2026 21:43:25 +0100 Subject: [PATCH 1/2] fix: release the compute log lock before rejecting a write to a finished capture A write that reached a capture already marked complete built its rejection while still holding the storage lock. Building it imports Python, which can release the GIL; the main thread then pruning the logs waited for that lock while holding the GIL, so neither could continue and the command never exited. The lock is now released before the rejection is raised. --- .../bindings/_helpers/compute_logs/storage.rs | 2 + .../sqlbuild/runtime/compute_logs/__init__.py | 0 .../runtime/compute_logs/_test_types.py | 14 ++ .../test_storage_lock_without_python.py | 130 ++++++++++++++++++ 4 files changed, 146 insertions(+) create mode 100644 tests/integration/src/sqlbuild/runtime/compute_logs/__init__.py create mode 100644 tests/integration/src/sqlbuild/runtime/compute_logs/_test_types.py create mode 100644 tests/integration/src/sqlbuild/runtime/compute_logs/test_storage_lock_without_python.py diff --git a/crates/sqlbuild-python/src/bindings/_helpers/compute_logs/storage.rs b/crates/sqlbuild-python/src/bindings/_helpers/compute_logs/storage.rs index 02bc5c5556..40b859097c 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/compute_logs/storage.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/compute_logs/storage.rs @@ -90,6 +90,7 @@ fn exact_count(value: &Bound<'_, PyAny>, fallback: i64) -> i64 { } impl LocalFilesystemComputeLogStorage { + /// Callers wait for this lock holding the GIL, so never run Python while holding it. fn locked(&self) -> PyResult> { self.state .lock() @@ -116,6 +117,7 @@ impl LocalFilesystemComputeLogStorage { ) -> PyResult<()> { let mut state = self.locked()?; let Some(writers) = state.writers.get_mut(invocation_id) else { + drop(state); return Err(not_active(py)); }; writers[stream].write_all(data)?; diff --git a/tests/integration/src/sqlbuild/runtime/compute_logs/__init__.py b/tests/integration/src/sqlbuild/runtime/compute_logs/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/integration/src/sqlbuild/runtime/compute_logs/_test_types.py b/tests/integration/src/sqlbuild/runtime/compute_logs/_test_types.py new file mode 100644 index 0000000000..d145089911 --- /dev/null +++ b/tests/integration/src/sqlbuild/runtime/compute_logs/_test_types.py @@ -0,0 +1,14 @@ +"""Test case types for compute log storage integration tests.""" + +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class StorageLockTestCase: + """A Python program making concurrent storage calls from two threads, and what it prints.""" + + description: str + script: str + expected_stdout: str diff --git a/tests/integration/src/sqlbuild/runtime/compute_logs/test_storage_lock_without_python.py b/tests/integration/src/sqlbuild/runtime/compute_logs/test_storage_lock_without_python.py new file mode 100644 index 0000000000..5e9370ac95 --- /dev/null +++ b/tests/integration/src/sqlbuild/runtime/compute_logs/test_storage_lock_without_python.py @@ -0,0 +1,130 @@ +"""Compute log storage never runs Python while holding its lock, so it cannot deadlock on the GIL.""" + +from __future__ import annotations + +import subprocess +import sys +from pathlib import Path +from textwrap import dedent + +import pytest + +from tests.integration.src.sqlbuild.runtime.compute_logs._test_types import ( + StorageLockTestCase, +) + +_DEADLOCK_TIMEOUT_SECONDS: int = 30 + +# A late write from a background thread reaches a capture that was already marked complete. +# Raising its rejection imports the storage exceptions, here forced to load afresh, and the +# import's file system calls release the GIL. The finder below holds that import open until the main thread is +# pruning, which takes the storage lock while holding the GIL. +_LATE_APPEND_DURING_PRUNE: str = dedent( + """ + import dataclasses + import sys + import threading + import time + from datetime import UTC, datetime + from pathlib import Path + + from sqlbuild.compute_logs import ( + COMPUTE_LOG_FORMAT_VERSION, + CaptureMetadata, + ComputeLogStream, + FinalCaptureMetadata, + LocalFilesystemComputeLogStorage, + ) + + EXCEPTIONS_MODULE = "sqlbuild.runtime.compute_logs.exceptions" + importing = threading.Event() + + + class SlowExceptionsImport: + def find_spec(self, name, path=None, target=None): + if name == EXCEPTIONS_MODULE and not importing.is_set(): + importing.set() + time.sleep(0.5) + return None + + + project_dir = Path(sys.argv[1]) + started_at = datetime.now(UTC) + metadata = CaptureMetadata( + format_version=COMPUTE_LOG_FORMAT_VERSION, + invocation_id="orders0000000000000000000000000a", + command="compile", + project_dir=str(project_dir), + started_at=started_at, + capture_date=started_at.date().isoformat(), + ) + storage = LocalFilesystemComputeLogStorage(project_dir=project_dir) + storage.start_capture(metadata) + storage.mark_complete( + FinalCaptureMetadata( + **dataclasses.asdict(metadata), + completed_at=datetime.now(UTC), + exit_code=1, + stdout_bytes=0, + stderr_bytes=0, + diagnostics_bytes=0, + ) + ) + rejected = [] + del sys.modules[EXCEPTIONS_MODULE] + sys.meta_path.insert(0, SlowExceptionsImport()) + + + def late_append(): + try: + storage.append( + invocation_id=metadata.invocation_id, + stream=ComputeLogStream.STDERR, + data=b"late customer export notice\\n", + ) + except Exception as error: + rejected.append(type(error).__name__) + + + writer = threading.Thread(target=late_append) + writer.start() + importing.wait() + storage.prune() + writer.join() + print(rejected) + """ +) + + +@pytest.mark.parametrize( + "test_case", + ( + StorageLockTestCase( + description="late append rejected while the main thread prunes", + script=_LATE_APPEND_DURING_PRUNE, + expected_stdout="['CaptureStateError']\n", + ), + ), + ids=lambda case: case.description, +) +def test_given_rejected_append_releasing_gil_when_pruning_then_both_threads_finish( + test_case: StorageLockTestCase, tmp_path: Path +) -> None: + try: + completed: subprocess.CompletedProcess[str] = subprocess.run( + [sys.executable, "-c", test_case.script, str(tmp_path)], + capture_output=True, + text=True, + timeout=_DEADLOCK_TIMEOUT_SECONDS, + check=False, + ) + except subprocess.TimeoutExpired: + pytest.fail(f"storage calls deadlocked for {_DEADLOCK_TIMEOUT_SECONDS}s") + + assert (completed.returncode, completed.stdout) == (0, test_case.expected_stdout), ( + completed.stderr + ) + + +if __name__ == "__main__": + pytest.main([__file__, "-vv"]) From c10b78caccc50c2e4bf95205f57760584f33835f Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Mon, 19 Oct 2026 21:43:25 +0100 Subject: [PATCH 2/2] test: fail a hung E2E command with every thread's stack instead of stalling the job run_sqb now gives each command ten minutes. A command still running then has its process group aborted with Python's fault handler enabled, so the failing test reports every thread's stack, and the group is killed. Standard input ends at once when a test passes none. --- Makefile | 3 +- .../cli/commands/shared/_test_types.py | 24 +++++ .../sqlbuild/cli/commands/shared/helpers.py | 78 +++++++++++++-- .../shared/test_run_with_hang_dump.py | 95 +++++++++++++++++++ 4 files changed, 193 insertions(+), 7 deletions(-) create mode 100644 tests/e2e/src/sqlbuild/cli/commands/shared/_test_types.py create mode 100644 tests/e2e/src/sqlbuild/cli/commands/shared/test_run_with_hang_dump.py diff --git a/Makefile b/Makefile index efba5d41b3..c1725a5a3d 100644 --- a/Makefile +++ b/Makefile @@ -179,7 +179,8 @@ E2E_DUCKDB_CLI_PATHS := \ tests/e2e/src/sqlbuild/cli/commands/main/snowflake \ tests/e2e/src/sqlbuild/cli/commands/main/sqlite \ tests/e2e/src/sqlbuild/cli/commands/main/sqlserver \ - tests/e2e/src/sqlbuild/cli/commands/main/test + tests/e2e/src/sqlbuild/cli/commands/main/test \ + tests/e2e/src/sqlbuild/cli/commands/shared E2E_DUCKDB_INSPECTION_PATHS := \ tests/e2e/src/sqlbuild/cli/commands/main/clone \ diff --git a/tests/e2e/src/sqlbuild/cli/commands/shared/_test_types.py b/tests/e2e/src/sqlbuild/cli/commands/shared/_test_types.py new file mode 100644 index 0000000000..08d61d09e7 --- /dev/null +++ b/tests/e2e/src/sqlbuild/cli/commands/shared/_test_types.py @@ -0,0 +1,24 @@ +"""Test case types for the shared E2E command helpers.""" + +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class HangDumpTestCase: + """A Python command that never finishes and what its hang report must show.""" + + description: str + script: str + expected_fragments: tuple[str, ...] + + +@dataclass(frozen=True) +class FinishedCommandTestCase: + """A Python command that finishes and the result the helper must return for it.""" + + description: str + script: str + input_text: str | None + expected_result: tuple[int, str, str] diff --git a/tests/e2e/src/sqlbuild/cli/commands/shared/helpers.py b/tests/e2e/src/sqlbuild/cli/commands/shared/helpers.py index 7b39be0dc4..d4cf1f3b99 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/shared/helpers.py +++ b/tests/e2e/src/sqlbuild/cli/commands/shared/helpers.py @@ -6,6 +6,8 @@ import os import pty import re +import shlex +import signal import struct import subprocess import termios @@ -22,6 +24,10 @@ SOURCE_LOADER_STRATEGIES_DIR: Path = ( REPO_ROOT / "tests" / "e2e" / "fixtures" / "source_loader_strategies" ) +# Far above any healthy command, so only a hang trips it: one test fails instead of the CI job. +SQB_COMMAND_TIMEOUT_SECONDS: float = 600.0 +_HANG_DUMP_GRACE_SECONDS: float = 30.0 +_HANG_OUTPUT_TAIL_CHARACTERS: int = 20_000 def prepare_waffle_shop(tmp_path: Path) -> Path: @@ -208,18 +214,78 @@ def run_sqb( process_env: dict[str, str] = dict(os.environ) process_env.update(env or {}) + process_env.setdefault("PYTHONFAULTHANDLER", "1") - return subprocess.run( - ["uv", "run", "sqb", "--project-dir", str(project_dir), *command], + return run_with_hang_dump( + argv=("uv", "run", "sqb", "--project-dir", str(project_dir), *command), cwd=working_dir or REPO_ROOT, - capture_output=True, - text=True, - input=input_text, env=process_env, - check=False, + input_text=input_text, + timeout_seconds=SQB_COMMAND_TIMEOUT_SECONDS, + ) + + +def run_with_hang_dump( + *, + argv: tuple[str, ...], + cwd: Path, + env: Mapping[str, str], + input_text: str | None, + timeout_seconds: float, +) -> subprocess.CompletedProcess[str]: + """Run `argv` to completion, or fail with every Python thread's stack once it hangs. + + Standard input carries `input_text`, or ends at once without it. The command runs in its own + process group. Past `timeout_seconds` the whole group receives + SIGABRT, so Python processes started with `PYTHONFAULTHANDLER` print all thread stacks to + stderr before they exit, and the group is then killed. + """ + + process: subprocess.Popen[str] = subprocess.Popen( + argv, + cwd=cwd, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + env=dict(env), + process_group=0, + ) + try: + stdout, stderr = process.communicate(input=input_text, timeout=timeout_seconds) + except subprocess.TimeoutExpired: + stdout, stderr = _dump_and_kill_group(process=process) + raise AssertionError( + f"{shlex.join(argv)} did not finish within {timeout_seconds:.0f}s; " + "its process group was aborted to dump thread stacks.\n" + f"--- stdout (last {_HANG_OUTPUT_TAIL_CHARACTERS} characters) ---\n" + f"{stdout[-_HANG_OUTPUT_TAIL_CHARACTERS:]}\n" + f"--- stderr (last {_HANG_OUTPUT_TAIL_CHARACTERS} characters) ---\n" + f"{stderr[-_HANG_OUTPUT_TAIL_CHARACTERS:]}" + ) from None + return subprocess.CompletedProcess( + args=list(argv), returncode=process.returncode, stdout=stdout, stderr=stderr ) +def _dump_and_kill_group(*, process: subprocess.Popen[str]) -> tuple[str, str]: + _signal_group(process_group=process.pid, signal_number=signal.SIGABRT) + try: + stdout, stderr = process.communicate(timeout=_HANG_DUMP_GRACE_SECONDS) + except subprocess.TimeoutExpired: + _signal_group(process_group=process.pid, signal_number=signal.SIGKILL) + stdout, stderr = process.communicate() + _signal_group(process_group=process.pid, signal_number=signal.SIGKILL) + return stdout or "", stderr or "" + + +def _signal_group(*, process_group: int, signal_number: signal.Signals) -> None: + try: + os.killpg(process_group, signal_number) + except ProcessLookupError: + return + + def run_sqb_with_pty( *, command: tuple[str, ...], diff --git a/tests/e2e/src/sqlbuild/cli/commands/shared/test_run_with_hang_dump.py b/tests/e2e/src/sqlbuild/cli/commands/shared/test_run_with_hang_dump.py new file mode 100644 index 0000000000..0a2052dbcb --- /dev/null +++ b/tests/e2e/src/sqlbuild/cli/commands/shared/test_run_with_hang_dump.py @@ -0,0 +1,95 @@ +"""A hung command fails its own test with every Python thread's stack instead of hanging the run.""" + +from __future__ import annotations + +import os +import subprocess +import sys +import time +from pathlib import Path + +import pytest + +from tests.e2e.src.sqlbuild.cli.commands.shared._test_types import ( + FinishedCommandTestCase, + HangDumpTestCase, +) +from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import run_with_hang_dump + + +@pytest.mark.parametrize( + "test_case", + [ + HangDumpTestCase( + description="main thread joins a worker blocked forever", + script=( + "import threading\n" + "def blocked_worker_wait():\n" + " threading.Event().wait()\n" + "def joining_main_thread():\n" + " worker = threading.Thread(target=blocked_worker_wait)\n" + " worker.start()\n" + " print('started', flush=True)\n" + " worker.join()\n" + "joining_main_thread()\n" + ), + expected_fragments=("started", "blocked_worker_wait", "joining_main_thread"), + ), + ], + ids=lambda case: case.description, +) +def test_given_hung_python_command_when_running_with_hang_dump_then_fails_with_thread_stacks( + test_case: HangDumpTestCase, tmp_path: Path +) -> None: + started: float = time.monotonic() + with pytest.raises(AssertionError) as raised: + run_with_hang_dump( + argv=(sys.executable, "-c", test_case.script), + cwd=tmp_path, + env={**os.environ, "PYTHONFAULTHANDLER": "1"}, + input_text=None, + timeout_seconds=2.0, + ) + + message: str = str(raised.value) + assert "did not finish within 2s" in message + assert [fragment in message for fragment in test_case.expected_fragments] == [True] * len( + test_case.expected_fragments + ), message + assert time.monotonic() - started < 30.0 + + +@pytest.mark.parametrize( + "test_case", + [ + FinishedCommandTestCase( + description="input reaches the command and its exit code and output return", + script="import sys; sys.stdout.write(sys.stdin.read().upper()); sys.exit(3)", + input_text="orders\n", + expected_result=(3, "ORDERS\n", ""), + ), + FinishedCommandTestCase( + description="without input the command reads end of input", + script="import sys; print(repr(sys.stdin.read()))", + input_text=None, + expected_result=(0, "''\n", ""), + ), + ], + ids=lambda case: case.description, +) +def test_given_finishing_command_when_running_with_hang_dump_then_returns_its_result( + test_case: FinishedCommandTestCase, tmp_path: Path +) -> None: + completed: subprocess.CompletedProcess[str] = run_with_hang_dump( + argv=(sys.executable, "-c", test_case.script), + cwd=tmp_path, + env=dict(os.environ), + input_text=test_case.input_text, + timeout_seconds=60.0, + ) + + assert (completed.returncode, completed.stdout, completed.stderr) == test_case.expected_result + + +if __name__ == "__main__": + pytest.main([__file__, "-vv"])