diff --git a/news/+run-shutdown-owner.bugfix.md b/news/+run-shutdown-owner.bugfix.md new file mode 100644 index 00000000000..4a7d1fd1d7e --- /dev/null +++ b/news/+run-shutdown-owner.bugfix.md @@ -0,0 +1 @@ +`reflex run` now stops the whole frontend process tree on SIGINT or SIGTERM, including a dev server left behind by `npm run`, a repeat right after the first signal no longer interrupts that cleanup, and a later Ctrl-C forces the exit when the cleanup is stuck. An exit the CLI asked for is no longer reported as a frontend failure. diff --git a/packages/reflex-base/news/+json-supervisor-bounded-shutdown.bugfix.md b/packages/reflex-base/news/+json-supervisor-bounded-shutdown.bugfix.md new file mode 100644 index 00000000000..d01b6970a5c --- /dev/null +++ b/packages/reflex-base/news/+json-supervisor-bounded-shutdown.bugfix.md @@ -0,0 +1 @@ +`reflex run --json` shutdown is bounded from the first SIGINT or SIGTERM: every signal is passed on to the app, the app's process tree is killed if it has not exited 15 seconds later, and a consumer that reads slowly or stops reading can no longer hang the supervisor or receive a cut JSON record. diff --git a/packages/reflex-base/src/reflex_base/utils/log.py b/packages/reflex-base/src/reflex_base/utils/log.py index 725313fe889..61d5830413f 100644 --- a/packages/reflex-base/src/reflex_base/utils/log.py +++ b/packages/reflex-base/src/reflex_base/utils/log.py @@ -26,12 +26,8 @@ import json import logging import os -import re import shutil -import signal -import subprocess import sys -import threading import time from collections.abc import Generator from pathlib import Path @@ -264,276 +260,6 @@ def _write_json(payload: dict, *, stderr: bool): stream.flush() -# The supervisor PID, set for the child of supervise_output() and inherited -# by its descendants. -_SUPERVISED_ENV_VAR = "REFLEX_OUTPUT_SUPERVISED" - -# How long a reader may sit idle after the child exits before the supervisor -# stops waiting for descendants that still hold the pipe. -_DRAIN_IDLE_SECONDS = 0.5 - -# Maximum drain time after the child exits, excluding time forwarding output. -_DRAIN_MAX_SECONDS = 5 - -# Stop shutdown after 30 seconds even when a consumer blocks every write. -_DRAIN_WALL_SECONDS = 30 - -# Line ends in child output; a lone ``\r`` ends a progress-bar update. -_LINE_END = re.compile(rb"\r\n|\r|\n") - - -def is_output_supervised() -> bool: - """Check whether this process runs under supervise_output(). - - Returns: - True if a parent process turns this process's output into JSON records. - """ - return bool(os.environ.get(_SUPERVISED_ENV_VAR)) - - -def _is_json_record(line: bytes) -> bool: - """Check whether a line already is a JSON log record. - - Args: - line: The line to check. - - Returns: - True if the line is a JSON object with a level and a message. - """ - if not line.startswith(b"{"): - return False - try: - record = json.loads(line) - except ValueError: - return False - return isinstance(record, dict) and "level" in record and "message" in record - - -def _to_records( - lines: list[bytes], traceback: list[str], level: str, name: str, final: bool -) -> bytes: - """Convert raw output lines to JSON lines. - - Lines that already are JSON log records pass through unchanged. A Python - traceback is collected, across calls, into one record with the traceback - text in its ``exception`` field. - - Args: - lines: Complete output lines, without their newline. - traceback: The lines of a traceback still being collected. - level: The record level for plain lines. - name: The logger name for the records. - final: Whether the output ended, so an open traceback is emitted too. - - Returns: - The JSON lines, each ending with a newline. - """ - timestamp = datetime.datetime.now(tz=datetime.UTC).isoformat() - out: list[bytes] = [] - - def record(message: str, level: str = level, exception: str | None = None): - payload = { - "timestamp": timestamp, - "level": level, - "logger": name, - "message": message, - } - if exception is not None: - payload["exception"] = exception - out.append(json.dumps(payload).encode()) - - for raw in lines: - if _is_json_record(raw): - out.append(raw) - continue - line = raw.decode("utf-8", "replace") - if traceback: - traceback.append(line) - # Frames are indented; the exception line is not. - if line.strip() and not line[0].isspace(): - record(line.strip(), "error", "\n".join(traceback) + "\n") - traceback.clear() - elif line.startswith("Traceback (most recent call last):"): - traceback.append(line) - else: - record(line) - if final and traceback: - record(traceback[-1].strip(), "error", "\n".join(traceback) + "\n") - return b"".join(line + b"\n" for line in out) - - -def _write_all(fd: int, data: bytes): - """Write all bytes to a file descriptor. - - Args: - fd: The file descriptor. - data: The bytes to write. - """ - view = memoryview(data) - while view: - view = view[os.write(fd, view) :] - - -class _OutputPump(threading.Thread): - """Reader thread that turns one child output pipe into JSON lines.""" - - def __init__(self, read_fd: int, write_fd: int, level: str, name: str): - """Create the reader. - - Args: - read_fd: The read end of the child's output pipe. - write_fd: The file descriptor to write JSON lines to. - level: The record level for plain lines. - name: The logger name for the records. - """ - super().__init__(name=f"reflex-output-{name}", daemon=True) - self.read_fd = read_fd - self.write_fd = write_fd - self.level = level - self.logger_name = name - # When the reader started waiting for data; None while it works. - self.idle_since: float | None = None - # Cumulative write time and the start of any ongoing write. - self._write_state: tuple[float, float | None] = (0.0, None) - self._write_lock = threading.Lock() - - def run(self): - """Forward the pipe until every writer closed it.""" - pending = b"" - traceback: list[str] = [] - while True: - self.idle_since = time.monotonic() - chunk = os.read(self.read_fd, 1 << 16) - self.idle_since = None - if not chunk: - break - data = pending + chunk - # A trailing \r may be the first half of a \r\n. - cut = len(data) - data.endswith(b"\r") - *lines, pending = _LINE_END.split(data[:cut]) - pending += data[cut:] - self._write(lines, traceback, final=False) - os.close(self.read_fd) - pending = pending.removesuffix(b"\r") - self._write([pending] if pending else [], traceback, final=True) - - def _write(self, lines: list[bytes], traceback: list[str], final: bool): - """Write converted lines, dropping them if the consumer is gone. - - Args: - lines: Complete output lines. - traceback: The lines of a traceback still being collected. - final: Whether the output ended. - """ - data = _to_records(lines, traceback, self.level, self.logger_name, final) - with self._write_lock: - elapsed, _ = self._write_state - started = time.monotonic() - self._write_state = (elapsed, started) - # Keep reading when the consumer is gone: a full pipe blocks the child. - with contextlib.suppress(OSError): - _write_all(self.write_fd, data) - with self._write_lock: - self._write_state = (elapsed + time.monotonic() - started, None) - - def writing_snapshot(self) -> tuple[float, float]: - """Capture the current time and cumulative write duration together. - - Returns: - The monotonic time and seconds spent writing up to that time. - """ - with self._write_lock: - elapsed, started = self._write_state - now = time.monotonic() - return now, elapsed if started is None else elapsed + now - started - - def drain(self, exited_at: float, writing_at_exit: float, wall_deadline: float): - """Wait until the pipe is drained after the child exited. - - Returns early when the reader sat idle for a while after the exit, or - when the drain took too long. Time spent forwarding output does not - count toward the active limit. A separate wall deadline bounds shutdown - even when a consumer stops reading. - - Args: - exited_at: The monotonic time of the pump's snapshot after child exit. - writing_at_exit: The cumulative write time at that snapshot. - wall_deadline: The shared monotonic deadline for shutdown. - """ - while self.is_alive(): - self.join(0.05) - now, writing_elapsed = self.writing_snapshot() - idle_since = self.idle_since - # Seconds charged against the five-second active drain budget. - drain_elapsed = now - exited_at - (writing_elapsed - writing_at_exit) - if ( - now > wall_deadline - or drain_elapsed > _DRAIN_MAX_SECONDS - or ( - idle_since is not None - and now - max(idle_since, exited_at) > _DRAIN_IDLE_SECONDS - ) - ): - return - - -def supervise_output(args: list[str]) -> int: - """Run a command and turn everything it and its descendants print into JSON records. - - The command runs with its stdout and stderr on pipes, which every process - it starts inherits. Reader threads write each line to this process's - stdout and stderr as a JSON record: lines that already are JSON log - records pass through, and a Python traceback becomes one record. - - Args: - args: The command to run. - - Returns: - The command's exit code. - """ - for stream in (sys.stdout, sys.stderr): - if stream is not None: - stream.flush() - # The readers decode UTF-8; Python otherwise uses the locale encoding. - # A configured error handler is kept. - _, _, errors = os.environ.get("PYTHONIOENCODING", "").partition(":") - env = { - "PYTHONUNBUFFERED": "1", - **os.environ, - _SUPERVISED_ENV_VAR: str(os.getpid()), - "PYTHONIOENCODING": f"utf-8:{errors}" if errors else "utf-8", - } - # Raw pipes that only the readers close: closing a pipe while a reader is - # blocked on it waits for that read on Windows. - out_read, out_write = os.pipe() - err_read, err_write = os.pipe() - proc = subprocess.Popen(args, stdout=out_write, stderr=err_write, env=env) - os.close(out_write) - os.close(err_write) - pumps = [ - _OutputPump(out_read, 1, "info", "stdout"), - _OutputPump(err_read, 2, "warning", "stderr"), - ] - for pump in pumps: - pump.start() - signal.signal(signal.SIGTERM, lambda *_: proc.terminate()) - while True: - try: - returncode = proc.wait() - break - except KeyboardInterrupt: - # A terminal Ctrl-C reaches the child too, a SIGINT sent only to - # this process does not. The child handles a repeated interrupt. - if sys.platform != "win32": - proc.send_signal(signal.SIGINT) - wall_deadline = time.monotonic() + _DRAIN_WALL_SECONDS - snapshots = [pump.writing_snapshot() for pump in pumps] - for pump, (exited_at, writing_at_exit) in zip(pumps, snapshots, strict=True): - pump.drain(exited_at, writing_at_exit, wall_deadline) - # A child killed by a signal reports -signum; shells report 128 + signum. - return returncode if returncode >= 0 else 128 - returncode - - class JsonHandler(logging.Handler): """Emit one JSON object per record for machine consumption.""" diff --git a/packages/reflex-base/src/reflex_base/utils/process_tree.py b/packages/reflex-base/src/reflex_base/utils/process_tree.py new file mode 100644 index 00000000000..750299dc14b --- /dev/null +++ b/packages/reflex-base/src/reflex_base/utils/process_tree.py @@ -0,0 +1,225 @@ +"""Stop a process together with everything it started. + +Reflex keeps the processes it starts in its own process group, so a terminal +Ctrl-C, a ``kill -- -`` from a service manager or a group SIGKILL reach +all of them at once. The price is that a signal sent to one PID stops only +that process: a package manager in front of a dev server, for example, exits +and leaves the server behind. :func:`stop_process_tree` closes that gap by +signalling every descendant it can find, waiting for all of them against one +deadline, and killing whatever is left. +""" + +from __future__ import annotations + +import contextlib +import os +import signal +import subprocess +import sys +import time +from collections.abc import Iterable, Iterator +from pathlib import Path + +# Linux lists every process under /proc; other POSIX systems need ps(1). +_PROC = Path("/proc") + + +def _stat_fields(pid: int | str) -> list[bytes] | None: + """Read the fixed fields of a process's ``/proc//stat``. + + Args: + pid: The process ID. + + Returns: + The fields after the command name (state, ppid, ...), or None when + the process is gone. + """ + try: + stat = (_PROC / str(pid) / "stat").read_bytes() + except OSError: + return None + # The command name may hold spaces or parentheses; the fixed fields start + # after its closing parenthesis. + return stat[stat.rfind(b")") + 1 :].split() + + +def _process_parents() -> Iterator[tuple[int, int]]: + """Yield ``(pid, ppid)`` for every process on a POSIX system. + + Yields: + Each process ID with its parent process ID. + """ + if sys.platform == "linux" and _PROC.is_dir(): + for entry in _PROC.iterdir(): + # A process that exited while being listed has no fields. + if entry.name.isdigit() and (fields := _stat_fields(entry.name)): + yield int(entry.name), int(fields[1]) + return + listing = subprocess.run( + ["ps", "-A", "-o", "pid=,ppid="], capture_output=True, text=True, check=False + ).stdout + for line in listing.splitlines(): + fields = line.split() + if len(fields) == 2: + yield int(fields[0]), int(fields[1]) + + +def descendant_pids(pid: int) -> list[int]: + """List every live descendant of a process, parents before their children. + + Args: + pid: The root process ID. + + Returns: + The descendant process IDs, or an empty list on Windows, where + ``taskkill /T`` walks the tree itself. + """ + if sys.platform == "win32": + return [] + children: dict[int, list[int]] = {} + for child, parent in _process_parents(): + children.setdefault(parent, []).append(child) + found: list[int] = [] + pending = [pid] + while pending: + own = children.get(pending.pop(), ()) + found.extend(own) + pending.extend(own) + return found + + +def _alive(pid: int) -> bool: + """Check whether a process still exists. + + Args: + pid: The process ID. + + Returns: + Whether the process still runs; an exited process nobody reaped yet + counts as gone. + """ + # An orphan is reparented to this process when it is PID 1 or a + # subreaper, in which case only this process can reap it. + with contextlib.suppress(ChildProcessError): + if os.waitpid(pid, os.WNOHANG)[0]: + return False + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + except PermissionError: + return True + if sys.platform == "linux": + fields = _stat_fields(pid) + return fields is not None and fields[0] != b"Z" + return True + + +def _signal(pids: list[int], signum: int): + """Send a signal to each process, ignoring the ones already gone. + + Args: + pids: The process IDs. + signum: The signal. + """ + for pid in pids: + with contextlib.suppress(ProcessLookupError, PermissionError): + os.kill(pid, signum) + + +def kill_descendants(pid: int, known: Iterable[int] = ()): + """SIGKILL every descendant of a process at once, without waiting. + + On Windows this does nothing: descendants are not listed there, and the + one caller acts on a console Ctrl-C, which the console delivers to every + process attached to it, so the tree already has the signal. + + Args: + pid: The root process ID, which is left running. + known: Process IDs seen below the root earlier. One still in the + root's process group is killed too, since a descendant whose + parent has exited is no longer found below the root. One that + left the group belongs to an unrelated process that took over + the ID, so it is spared. + """ + if sys.platform == "win32": + return + group = _group_of(pid) + pids = [p for p in known if _group_of(p) == group] + _signal([*pids, *descendant_pids(pid)], signal.SIGKILL) + + +def _group_of(pid: int) -> int | None: + """Look up a process's process group. + + Args: + pid: The process ID. + + Returns: + The process group ID, or None when the process is gone or not ours + to ask about. + """ + try: + return os.getpgid(pid) + except (ProcessLookupError, PermissionError): + return None + + +def _wait(process: subprocess.Popen, descendants: list[int], deadline: float): + """Wait until the process and its descendants are gone or the deadline passes. + + Args: + process: The process, which is reaped here. + descendants: The descendant process IDs to wait for. + deadline: The monotonic time to give up at. + """ + while True: + remaining = deadline - time.monotonic() + with contextlib.suppress(subprocess.TimeoutExpired): + process.wait(timeout=max(remaining, 0)) + if process.returncode is not None and not any(map(_alive, descendants)): + return + if remaining <= 0: + return + time.sleep(min(0.02, remaining)) + + +def stop_process_tree(process: subprocess.Popen, timeout: float = 5.0): + """Stop a process and every process below it. + + On POSIX the process and its descendants get SIGTERM, then SIGKILL if any + of them is still alive after ``timeout`` seconds; a descendant started + during the wait is caught by a second look before the kill, as long as + the process itself still runs. A process that forks while handling + SIGTERM and exits before that fork is seen leaves it behind: without a + process group of its own, descendant discovery is best effort. On Windows + ``taskkill /T /F`` ends the whole tree at once. The process is reaped + either way, so ``process.returncode`` is set afterwards. + + Args: + process: The process to stop. + timeout: Seconds to wait for a graceful exit before killing. + """ + if sys.platform == "win32": + if process.poll() is None: + subprocess.run( + ["taskkill", "/T", "/F", "/PID", str(process.pid)], + capture_output=True, + check=False, + ) + with contextlib.suppress(subprocess.TimeoutExpired): + process.wait(timeout=timeout) + return + if process.poll() is not None: + return + descendants = descendant_pids(process.pid) + _signal([process.pid, *descendants], signal.SIGTERM) + _wait(process, descendants, time.monotonic() + timeout) + if process.returncode is None: + descendants = [*descendants, *descendant_pids(process.pid)] + _signal([process.pid], signal.SIGKILL) + survivors = [pid for pid in descendants if _alive(pid)] + if survivors: + _signal(survivors, signal.SIGKILL) + # A killed process disappears promptly; a short wait reaps the child. + _wait(process, survivors, time.monotonic() + 1) diff --git a/packages/reflex-base/src/reflex_base/utils/supervisor.py b/packages/reflex-base/src/reflex_base/utils/supervisor.py new file mode 100644 index 00000000000..f2573b24d34 --- /dev/null +++ b/packages/reflex-base/src/reflex_base/utils/supervisor.py @@ -0,0 +1,447 @@ +"""Output supervisor for ``reflex run --json``. + +:func:`supervise_output` re-runs the command with its stdout and stderr on +pipes and turns every line into a JSON record, so a consumer sees one JSON +object per line even for output that never went through ``logging``. It also +owns the command's shutdown: signals are passed on, the command tree is +killed after a deadline, and the drain never leaves a cut record on a pipe. +""" + +from __future__ import annotations + +import contextlib +import datetime +import json +import os +import re +import select +import signal +import subprocess +import sys +import threading +import time +from typing import TYPE_CHECKING + +from reflex_base.utils.process_tree import stop_process_tree + +if TYPE_CHECKING: + from collections.abc import Iterator + from types import FrameType + + +# The supervisor PID, set for the child of supervise_output() and inherited +# by its descendants. +_SUPERVISED_ENV_VAR = "REFLEX_OUTPUT_SUPERVISED" + +# How long a reader may sit idle after the child exits before the supervisor +# stops waiting for descendants that still hold the pipe. +_DRAIN_IDLE_SECONDS = 0.5 + +# Maximum drain time after the child exits, excluding time forwarding output. +_DRAIN_MAX_SECONDS = 5 + +# Seconds after the first shutdown signal before the command tree is killed. +_STOP_DEADLINE_SECONDS = 15 + +# Stop shutdown after 30 seconds even when a consumer blocks every write. +_DRAIN_WALL_SECONDS = 30 + +# Seconds a reader gets to finish the record it is writing when the drain ends. +_DRAIN_RECORD_GRACE_SECONDS = 2 + +# Line ends in child output; a lone ``\r`` ends a progress-bar update. +_LINE_END = re.compile(rb"\r\n|\r|\n") + +# Largest write a POSIX pipe accepts whole or not at all. Windows has no such +# constant and no such guarantee; the Linux value keeps writes small there. +_PIPE_BUF = getattr(select, "PIPE_BUF", 4096) + + +def is_output_supervised() -> bool: + """Check whether this process runs under supervise_output(). + + Returns: + True if a parent process turns this process's output into JSON records. + """ + return bool(os.environ.get(_SUPERVISED_ENV_VAR)) + + +def _is_json_record(line: bytes) -> bool: + """Check whether a line already is a JSON log record. + + Args: + line: The line to check. + + Returns: + True if the line is a JSON object with a level and a message. + """ + if not line.startswith(b"{"): + return False + try: + record = json.loads(line) + except ValueError: + return False + return isinstance(record, dict) and "level" in record and "message" in record + + +def _to_records( + lines: list[bytes], traceback: list[str], level: str, name: str, final: bool +) -> list[bytes]: + """Convert raw output lines to JSON lines. + + Lines that already are JSON log records pass through unchanged. A Python + traceback is collected, across calls, into one record with the traceback + text in its ``exception`` field. + + Args: + lines: Complete output lines, without their newline. + traceback: The lines of a traceback still being collected. + level: The record level for plain lines. + name: The logger name for the records. + final: Whether the output ended, so an open traceback is emitted too. + + Returns: + The JSON records, each ending with a newline. + """ + timestamp = datetime.datetime.now(tz=datetime.UTC).isoformat() + out: list[bytes] = [] + + def record(message: str, level: str = level, exception: str | None = None): + payload = { + "timestamp": timestamp, + "level": level, + "logger": name, + "message": message, + } + if exception is not None: + payload["exception"] = exception + out.append(json.dumps(payload).encode()) + + for raw in lines: + if _is_json_record(raw): + out.append(raw) + continue + line = raw.decode("utf-8", "replace") + if traceback: + traceback.append(line) + # Frames are indented; the exception line is not. + if line.strip() and not line[0].isspace(): + record(line.strip(), "error", "\n".join(traceback) + "\n") + traceback.clear() + elif line.startswith("Traceback (most recent call last):"): + traceback.append(line) + else: + record(line) + if final and traceback: + record(traceback[-1].strip(), "error", "\n".join(traceback) + "\n") + return [line + b"\n" for line in out] + + +def _atomic_chunks(records: list[bytes]) -> Iterator[bytes]: + """Group records into writes no larger than the pipe's atomic write size. + + A write of at most ``PIPE_BUF`` bytes enters a POSIX pipe whole or not at + all, so a reader thread that is abandoned mid-write can never leave a + partial record behind; only a single record above that size can be cut. + + Args: + records: The JSON records, each ending with a newline. + + Yields: + Chunks of consecutive records. + """ + chunk: list[bytes] = [] + size = 0 + for record in records: + if chunk and size + len(record) > _PIPE_BUF: + yield b"".join(chunk) + chunk, size = [], 0 + chunk.append(record) + size += len(record) + if chunk: + yield b"".join(chunk) + + +def _write_all(fd: int, data: bytes): + """Write all bytes to a file descriptor. + + Args: + fd: The file descriptor. + data: The bytes to write. + """ + view = memoryview(data) + while view: + view = view[os.write(fd, view) :] + + +class _OutputPump(threading.Thread): + """Reader thread that turns one child output pipe into JSON lines.""" + + def __init__(self, read_fd: int, write_fd: int, level: str, name: str): + """Create the reader. + + Args: + read_fd: The read end of the child's output pipe. + write_fd: The file descriptor to write JSON lines to. + level: The record level for plain lines. + name: The logger name for the records. + """ + super().__init__(name=f"reflex-output-{name}", daemon=True) + self.read_fd = read_fd + self.write_fd = write_fd + self.level = level + self.logger_name = name + # Set by drain() when time is up: finish the current write, drop the rest. + self._dropping = False + # When the reader started waiting for data; None while it works. + self.idle_since: float | None = None + # Cumulative write time and the start of any ongoing write. + self._write_state: tuple[float, float | None] = (0.0, None) + self._write_lock = threading.Lock() + # Clear while a write is in progress. + self._write_done = threading.Event() + self._write_done.set() + + def run(self): + """Forward the pipe until every writer closed it.""" + pending = b"" + traceback: list[str] = [] + while True: + self.idle_since = time.monotonic() + chunk = os.read(self.read_fd, 1 << 16) + self.idle_since = None + if not chunk: + break + data = pending + chunk + # A trailing \r may be the first half of a \r\n. + cut = len(data) - data.endswith(b"\r") + *lines, pending = _LINE_END.split(data[:cut]) + pending += data[cut:] + self._write(lines, traceback, final=False) + os.close(self.read_fd) + pending = pending.removesuffix(b"\r") + self._write([pending] if pending else [], traceback, final=True) + + def _write(self, lines: list[bytes], traceback: list[str], final: bool): + """Write converted lines, dropping them if the consumer is gone. + + Args: + lines: Complete output lines. + traceback: The lines of a traceback still being collected. + final: Whether the output ended. + """ + records = _to_records(lines, traceback, self.level, self.logger_name, final) + if not records or self._dropping: + return + self._write_done.clear() + with self._write_lock: + elapsed, _ = self._write_state + started = time.monotonic() + self._write_state = (elapsed, started) + # Keep reading when the consumer is gone: a full pipe blocks the child. + with contextlib.suppress(OSError): + for chunk in _atomic_chunks(records): + if self._dropping: + break + _write_all(self.write_fd, chunk) + with self._write_lock: + self._write_state = (elapsed + time.monotonic() - started, None) + self._write_done.set() + + def writing_snapshot(self) -> tuple[float, float]: + """Capture the current time and cumulative write duration together. + + Returns: + The monotonic time and seconds spent writing up to that time. + """ + with self._write_lock: + elapsed, started = self._write_state + now = time.monotonic() + return now, elapsed if started is None else elapsed + now - started + + def drain(self, exited_at: float, writing_at_exit: float, wall_deadline: float): + """Wait until the pipe is drained after the child exited. + + Returns early when the reader sat idle for a while after the exit, or + when the drain took too long. Time spent forwarding output does not + count toward the active limit. A separate wall deadline bounds shutdown + even when a consumer stops reading. A write in progress when time is + up gets a short grace to complete; writes never exceed the pipe's + atomic size, so an abandoned one cannot leave a partial record. + + Args: + exited_at: The monotonic time of the pump's snapshot after child exit. + writing_at_exit: The cumulative write time at that snapshot. + wall_deadline: The shared monotonic deadline for shutdown. + """ + while self.is_alive(): + self.join(0.05) + now, writing_elapsed = self.writing_snapshot() + idle_since = self.idle_since + # Seconds charged against the five-second active drain budget. + drain_elapsed = now - exited_at - (writing_elapsed - writing_at_exit) + if ( + now > wall_deadline + or drain_elapsed > _DRAIN_MAX_SECONDS + or ( + idle_since is not None + and now - max(idle_since, exited_at) > _DRAIN_IDLE_SECONDS + ) + ): + break + self._dropping = True + self._write_done.wait(_DRAIN_RECORD_GRACE_SECONDS) + + +def supervise_output(args: list[str]) -> int: + """Run a command and turn everything it and its descendants print into JSON records. + + The command runs with its stdout and stderr on pipes, which every process + it starts inherits. Reader threads write each line to this process's + stdout and stderr as a JSON record: lines that already are JSON log + records pass through, and a Python traceback becomes one record. + + Shutdown is bounded from the first SIGINT or SIGTERM: each one is passed + on to the command, its whole process tree is killed if it is still + running ``_STOP_DEADLINE_SECONDS`` later, and the output drain ends by + ``_DRAIN_WALL_SECONDS`` after the request. Output goes out in writes of + at most ``PIPE_BUF`` bytes, so on a pipe a record up to that size is + never cut; a larger one can be when its consumer is slower than the + grace the drain allows. + + Args: + args: The command to run. + + Returns: + The command's exit code. + """ + for stream in (sys.stdout, sys.stderr): + if stream is not None: + stream.flush() + # The readers decode UTF-8; Python otherwise uses the locale encoding. + # A configured error handler is kept. + _, _, errors = os.environ.get("PYTHONIOENCODING", "").partition(":") + env = { + "PYTHONUNBUFFERED": "1", + **os.environ, + _SUPERVISED_ENV_VAR: str(os.getpid()), + "PYTHONIOENCODING": f"utf-8:{errors}" if errors else "utf-8", + } + # Raw pipes that only the readers close: closing a pipe while a reader is + # blocked on it waits for that read on Windows. + out_read, out_write = os.pipe() + err_read, err_write = os.pipe() + # The command stays in this process group: a terminal Ctrl-C or a group + # kill from a service manager reaches it and its workers directly. + proc = subprocess.Popen(args, stdout=out_write, stderr=err_write, env=env) + os.close(out_write) + os.close(err_write) + shutdown = _Shutdown(proc) + pumps = [ + _OutputPump(out_read, 1, "info", "stdout"), + _OutputPump(err_read, 2, "warning", "stderr"), + ] + for pump in pumps: + pump.start() + # SIGTERM only on Windows, where Ctrl-C arrives as KeyboardInterrupt. + handled = ( + (signal.SIGTERM, signal.SIGINT) if os.name == "posix" else (signal.SIGTERM,) + ) + previous = {signum: signal.signal(signum, shutdown.on_signal) for signum in handled} + try: + while True: + try: + returncode = proc.wait() + break + except KeyboardInterrupt: + # The command shares the console and received the Ctrl-C too. + shutdown.request(None) + requested_at = shutdown.requested_at + _drain_pumps( + pumps, + (requested_at if requested_at is not None else time.monotonic()) + + _DRAIN_WALL_SECONDS, + ) + # A child killed by a signal reports -signum; shells report 128 + signum. + return returncode if returncode >= 0 else 128 - returncode + finally: + shutdown.cancel() + for signum, handler in previous.items(): + signal.signal(signum, handler) + + +def _drain_pumps(pumps: list[_OutputPump], wall_deadline: float): + """Drain the readers after the command exited. + + Args: + pumps: The reader threads. + wall_deadline: The monotonic time by which the drain ends. + """ + snapshots = [pump.writing_snapshot() for pump in pumps] + for pump, (exited_at, writing_at_exit) in zip(pumps, snapshots, strict=True): + while True: + try: + pump.drain(exited_at, writing_at_exit, wall_deadline) + break + except KeyboardInterrupt: + # A repeated Windows Ctrl-C does not cut the drain short. + continue + + +class _Shutdown: + """Bounded shutdown of the supervised command. + + Every shutdown signal is passed on to the command, which decides what a + repeat means. The first one also starts a deadline after which the whole + command tree is killed, so a command that never finishes stopping, or one + blocked on a full output pipe, cannot keep the supervisor alive. + """ + + def __init__(self, proc: subprocess.Popen): + """Create the shutdown tracker. + + Args: + proc: The supervised command. + """ + self.proc = proc + self.requested_at: float | None = None + self._timer: threading.Timer | None = None + + def on_signal(self, signum: int, _frame: FrameType | None): + """Handle a shutdown signal. + + Args: + signum: The signal number. + _frame: The interrupted frame. + """ + self.request(signum) + + def request(self, signum: int | None): + """Pass a shutdown request on to the command and arm the deadline. + + Args: + signum: The signal to forward, or None when the command already + received it (a Windows console Ctrl-C). + """ + if self.requested_at is None: + self.requested_at = time.monotonic() + self._timer = threading.Timer(_STOP_DEADLINE_SECONDS, self._kill) + self._timer.daemon = True + self._timer.start() + if signum is None: + return + if os.name == "posix": + self.proc.send_signal(signum) + else: + # Windows has no graceful signal: terminating only the command + # would leave its workers behind, so the whole tree goes at once. + self._kill() + + def _kill(self): + """Stop the command tree once the deadline passed.""" + stop_process_tree(self.proc, timeout=1) + + def cancel(self): + """Disarm the deadline once the command exited.""" + if self._timer is not None: + self._timer.cancel() diff --git a/pyproject.toml b/pyproject.toml index 8d59db5a773..1eef5b8d253 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,7 +32,7 @@ dependencies = [ "starlette >=1.3.1", "typing_extensions >=4.13.0", "wrapt >=1.17.0,<2.6", - "reflex-base >= 0.10.1", + "reflex-base >= 0.10.2.dev0", "reflex-components-code >= 0.10.1", "reflex-components-core >= 0.10.0", "reflex-components-dataeditor >= 0.10.1", diff --git a/reflex/reflex.py b/reflex/reflex.py index 08384e1ff56..c4ba578d204 100644 --- a/reflex/reflex.py +++ b/reflex/reflex.py @@ -3,10 +3,13 @@ from __future__ import annotations import logging +import os +import select import signal import sys -from collections.abc import Callable -from contextlib import contextmanager, nullcontext +import time +from collections.abc import Callable, Generator +from contextlib import contextmanager, nullcontext, suppress from importlib import import_module from importlib.util import find_spec from pathlib import Path @@ -17,7 +20,12 @@ from reflex_base import constants from reflex_base.config import get_config, reload_config from reflex_base.environment import environment -from reflex_base.utils import console, log +from reflex_base.utils import console, log, supervisor +from reflex_base.utils.process_tree import ( + descendant_pids, + kill_descendants, + stop_process_tree, +) from reflex.utils.cli_options import log_options @@ -406,21 +414,81 @@ def _compile_app_worker( otel.flush() -@contextmanager -def _frontend_sigterm_handler(enabled: bool): - """Exit cleanly on SIGTERM for a frontend-only run.""" - if not enabled: - yield - return +# Seconds after a stop request in which a repeat is the same request delivered +# twice; a later repeat forces the exit. +_REPEAT_IGNORE_SECONDS = 1 + +# Windows never delivers SIGTERM, but Ctrl-C arrives as SIGINT there too. +_SHUTDOWN_SIGNALS = (signal.SIGINT, signal.SIGTERM) + + +def _write_stderr_now(message: bytes) -> None: + """Write to stderr unless that would block. - def stop_frontend(signum: int, frame: FrameType | None) -> None: - raise SystemExit(0) + A forced exit must not wait on a pipe nobody reads. On Windows ``select`` + takes only sockets, so the write is attempted directly there. - previous = signal.signal(signal.SIGTERM, stop_frontend) + Args: + message: The bytes to write. + """ + with suppress(OSError): + if sys.platform == "win32" or select.select([], [2], [], 0)[1]: + os.write(2, message) + + +@contextmanager +def _shutdown_signals() -> Generator[Callable[[], None], None, None]: + """Turn the first SIGINT or SIGTERM into a clean unwind; force on a later one. + + The first signal raises in the main thread so the run body unwinds into + its cleanup. A repeat within ``_REPEAT_IGNORE_SECONDS``, such as the + second delivery a terminal Ctrl-C produces next to the one the output + supervisor forwards, is ignored so it cannot interrupt that cleanup. A + later repeat means the cleanup is stuck: every descendant is killed and + the process exits at once (on Windows only the exit: a console Ctrl-C + already reached every process on the console). The backend server + installs its own handlers while it serves and does not restore these, so + the cleanup puts them back. + + Yields: + A function the run body calls when its cleanup begins, so a signal + arriving only then counts as a repeat. + """ + stopping_at: float | None = None + # Everything below this process when the cleanup began. A forced exit + # kills these as well: a server whose wrapper exited during the cleanup + # is no longer found below this process. + tree: list[int] = [] + + def begin_cleanup() -> None: + nonlocal stopping_at + if stopping_at is None: + stopping_at = time.monotonic() + tree.extend(descendant_pids(os.getpid())) + # The backend server installs its own handlers and does not restore them. + for sig in _SHUTDOWN_SIGNALS: + signal.signal(sig, stop) + + def stop(signum: int, frame: FrameType | None) -> None: + if stopping_at is None: + begin_cleanup() + if signum == signal.SIGINT: + raise KeyboardInterrupt + raise SystemExit(0) + if time.monotonic() - stopping_at < _REPEAT_IGNORE_SECONDS: + return + # A normal exit would wait for the thread a stuck cleanup blocks, for + # example one reading a pipe that a process outside the tree holds open. + kill_descendants(os.getpid(), tree) + _write_stderr_now(b"Forced exit: the cleanup did not finish.\n") + os._exit(128 + signum) + + previous = {sig: signal.signal(sig, stop) for sig in _SHUTDOWN_SIGNALS} try: - yield + yield begin_cleanup finally: - signal.signal(signal.SIGTERM, previous) + for sig, handler in previous.items(): + signal.signal(sig, handler) def _run_dev( @@ -472,14 +540,15 @@ def _run_dev( def stop_frontend() -> None: with exec._frontend_process_lock: exec._frontend_shutting_down = True - if (process := exec.frontend_process) is not None: - if process.poll() is None: - process.terminate() - exec.frontend_process = None + process, exec.frontend_process = exec.frontend_process, None + if process is not None: + # The package manager in front of the dev server does not always + # pass a signal on, so the whole tree below it is stopped. + stop_process_tree(process, timeout=exec._FRONTEND_STOP_TIMEOUT) try: with ( - _frontend_sigterm_handler(frontend_only and sys.platform != "win32"), + _shutdown_signals() as begin_cleanup, processes.run_concurrently_context( *commands, interrupt_on_failure=not frontend_only ) as tasks, @@ -494,11 +563,8 @@ def stop_frontend() -> None: config.loglevel.subprocess_level(), running_mode.has_frontend(), ) - # The windows uvicorn bug workaround - # https://github.com/reflex-dev/reflex/issues/2335 - if constants.IS_WINDOWS and exec.frontend_process: - exec.kill(exec.frontend_process.pid) finally: + begin_cleanup() stop_frontend() finally: stop_frontend() @@ -788,11 +854,11 @@ def run( """Run the app in the current directory.""" from reflex.utils import prerequisites - if log.is_json_mode() and not log.is_output_supervised(): + if log.is_json_mode() and not supervisor.is_output_supervised(): # Run the command again below a process that turns every line it and # its workers print into a JSON record. raise SystemExit( - log.supervise_output([sys.executable, "-m", "reflex", *sys.argv[1:]]) + supervisor.supervise_output([sys.executable, "-m", "reflex", *sys.argv[1:]]) ) if frontend_only and backend_only: diff --git a/reflex/utils/exec.py b/reflex/utils/exec.py index 5c4a8fb9726..d9c9533d8a6 100644 --- a/reflex/utils/exec.py +++ b/reflex/utils/exec.py @@ -26,6 +26,7 @@ from reflex_base.telemetry_context import CompileTrigger from reflex_base.utils import console, log from reflex_base.utils.decorator import once +from reflex_base.utils.process_tree import stop_process_tree from reflex.utils import path_ops from reflex.utils.misc import get_module_path @@ -33,10 +34,12 @@ logger = logging.getLogger(__name__) -# For uvicorn windows bug fix (#2335) +# The frontend dev server process, stopped by `reflex run` when it unwinds. frontend_process = None _frontend_process_lock = threading.Lock() _frontend_shutting_down = False +# Seconds the frontend tree gets to exit on SIGTERM before SIGKILL. +_FRONTEND_STOP_TIMEOUT = 5 DEV_BACKEND_RELOAD_MARKER = ".reflex_dev_backend_started" @@ -178,25 +181,6 @@ def get_changes(old: dict[str, str], new: dict[str, str]) -> Change: return dependencies_change, dev_dependencies_change -def kill(proc_pid: int): - """Kills a process and all its child processes. - - Requires the `psutil` library to be installed. - - Args: - proc_pid: The process ID of the process to be killed. - - Example: - >>> kill(1234) - """ - import psutil - - process = psutil.Process(proc_pid) - for proc in process.children(recursive=True): - proc.kill() - process.kill() - - def notify_frontend(url: str, backend_present: bool): """Output a string notifying where the frontend is running. @@ -300,10 +284,15 @@ def run_process_and_launch_url( global frontend_process with _frontend_process_lock: frontend_process = process - if _frontend_shutting_down and process.poll() is None: - process.terminate() + launched_too_late = _frontend_shutting_down + if launched_too_late: + stop_process_tree(process, timeout=_FRONTEND_STOP_TIMEOUT) if process.stdout: - for line in processes.stream_logs("Starting frontend", process): + for line in processes.stream_logs( + "Starting frontend", + process, + expected_exit=lambda: _frontend_shutting_down, + ): new_content, new_hash = get_package_json_and_hash(json_file_path) if new_hash != last_hash: dependencies_change, dev_dependencies_change = ( diff --git a/reflex/utils/processes.py b/reflex/utils/processes.py index 2e5d449c582..270dd8fd7b6 100644 --- a/reflex/utils/processes.py +++ b/reflex/utils/processes.py @@ -402,6 +402,7 @@ def stream_logs( suppress_errors: bool = False, analytics_enabled: bool = False, prior_logs: tuple[tuple[str, ...], ...] = (), + expected_exit: Callable[[], bool] | None = None, ): """Stream the logs for a process. @@ -412,6 +413,8 @@ def stream_logs( suppress_errors: If True, do not exit if errors are encountered (for fallback). analytics_enabled: Whether analytics are enabled for this command. prior_logs: The logs of the prior processes that have been run. + expected_exit: Reports whether the process was asked to stop, in which + case its exit code is not a failure whatever its value. Yields: The lines of the process output. @@ -463,7 +466,11 @@ def stream_logs( *(-sig for sig in interrupt_signals), *(128 + sig for sig in interrupt_signals), } - if process.returncode not in accepted_return_codes and not suppress_errors: + if ( + process.returncode not in accepted_return_codes + and not suppress_errors + and not (expected_exit is not None and expected_exit()) + ): logger.error(f"{message} failed with exit code {process.returncode}") if "".join(logs).count("CERT_HAS_EXPIRED") > 0: bunfig = prerequisites.get_web_dir() / constants.Bun.CONFIG_PATH diff --git a/tests/integration/test_run_shutdown.py b/tests/integration/test_run_shutdown.py new file mode 100644 index 00000000000..1375379afef --- /dev/null +++ b/tests/integration/test_run_shutdown.py @@ -0,0 +1,341 @@ +"""Stop a real ``reflex run`` every way a user or a service manager does. + +Each case starts the CLI on a small app, waits until the app is up, delivers +one kind of stop and checks the same three things: the CLI exits promptly, +nothing it started is left running, and the output holds no error. The +cases differ only in the route the stop takes: a terminal Ctrl-C reaches the +whole process group, ``kill`` and service managers reach the CLI's PID only, +and ``--json`` puts the output supervisor in front of everything. +""" + +from __future__ import annotations + +import contextlib +import json +import os +import shutil +import signal +import socket +import subprocess +import sys +import textwrap +import threading +import time +from collections.abc import Generator +from pathlib import Path + +import pytest +from reflex_base.utils.process_tree import descendant_pids + +pytestmark = pytest.mark.skipif(sys.platform == "win32", reason="POSIX signal delivery") + +# Seconds for the app to come up, including the first frontend install. +START_TIMEOUT = 600 +# Seconds the CLI gets to exit after a stop request. +STOP_TIMEOUT = 20 + +APP_SOURCE = """ +import reflex as rx + + +class State(rx.State): + count: int = 0 + + def increment(self): + self.count += 1 + + +def index(): + return rx.vstack(rx.text(State.count), rx.button("+", on_click=State.increment)) + + +app = rx.App() +app.add_page(index) +""" + + +@pytest.fixture(scope="module") +def app_dir(tmp_path_factory: pytest.TempPathFactory) -> Path: + """Create a minimal app and initialize its frontend once. + + Args: + tmp_path_factory: The pytest temporary path factory. + + Returns: + The app directory. + """ + root = tmp_path_factory.mktemp("run_shutdown") + (root / "rxconfig.py").write_text( + 'import reflex as rx\n\nconfig = rx.Config(app_name="shutdown_app")\n' + ) + package = root / "shutdown_app" + package.mkdir() + (package / "__init__.py").write_text("") + (package / "shutdown_app.py").write_text(textwrap.dedent(APP_SOURCE)) + subprocess.run( + [sys.executable, "-m", "reflex", "init", "--loglevel", "warning"], + cwd=root, + check=True, + timeout=START_TIMEOUT, + ) + return root + + +def _free_port() -> int: + """Pick a free TCP port. + + Returns: + A port that was free when checked. + """ + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] + + +def _alive(pid: int) -> bool: + """Check whether a process still runs. + + Args: + pid: The process ID. + + Returns: + Whether the process exists and is not a zombie. + """ + try: + os.kill(pid, 0) + if sys.platform == "linux": + stat = Path(f"/proc/{pid}/stat").read_bytes() + return stat.rsplit(b")", 1)[1].split()[0] != b"Z" + except OSError: + return False + return True + + +def _describe(pid: int) -> str: + """Describe a process for a failure message. + + Args: + pid: The process ID. + + Returns: + The PID with its command line where available. + """ + try: + command = Path(f"/proc/{pid}/cmdline").read_bytes() + except OSError: + return str(pid) + text = command.replace(b"\0", b" ").decode(errors="replace") + return f"{pid} {text[:80]}" + + +class _RunningApp: + """A ``reflex run`` started in its own process group, with its output.""" + + def __init__(self, app_dir: Path, json_mode: bool, npm: bool): + """Start the CLI. + + Args: + app_dir: The app directory. + json_mode: Whether to run with ``--json``. + npm: Whether to put npm instead of bun in front of the dev server. + """ + env = {**os.environ, "REFLEX_USE_NPM": "1" if npm else ""} + if json_mode: + env["REFLEX_LOG_JSON"] = "true" + self.process = subprocess.Popen( + [ + sys.executable, + "-m", + "reflex", + "run", + "--frontend-port", + str(_free_port()), + "--backend-port", + str(_free_port()), + ], + cwd=app_dir, + env=env, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + start_new_session=True, + ) + self.lines: list[bytes] = [] + self.running = threading.Event() + self.json_mode = json_mode + self.descendants: list[int] = [] + # Read continuously: a full pipe would block the app during shutdown, + # and EOF only arrives once every process holding the pipe is gone. + self.reader = threading.Thread(target=self._read, daemon=True) + self.reader.start() + + def _read(self): + """Collect output lines and flag the running banner.""" + assert self.process.stdout is not None + for line in self.process.stdout: + self.lines.append(line) + if b"App running at" in line: + self.running.set() + + @property + def output(self) -> str: + """The output collected so far. + + Returns: + The decoded output. + """ + return b"".join(self.lines).decode("utf-8", "replace") + + def wait_until_running(self): + """Wait until the app reports it is running.""" + deadline = time.monotonic() + START_TIMEOUT + while not self.running.wait(0.5): + if self.process.poll() is not None: + pytest.fail(f"reflex run exited early:\n{self.output}") + if time.monotonic() > deadline: + pytest.fail(f"reflex run did not start in time:\n{self.output}") + # Give the backend and frontend trees a moment to settle. + time.sleep(2) + self.descendants = descendant_pids(self.process.pid) + assert self.descendants, "reflex run started no child processes" + + def stop(self, deliver) -> float: + """Deliver a stop request and wait for the CLI to exit. + + Args: + deliver: Sends the stop given the CLI process. + + Returns: + Seconds between the request and the exit. + """ + started = time.monotonic() + deliver(self.process) + try: + self.process.wait(timeout=STOP_TIMEOUT) + except subprocess.TimeoutExpired: + alive = [_describe(pid) for pid in self.descendants if _alive(pid)] + self.kill() + pytest.fail( + f"reflex run did not exit after the stop request; still running: " + f"{alive}\n{self.output[-3000:]}" + ) + elapsed = time.monotonic() - started + # The pipe closes once every descendant is gone; stragglers are killed + # by the fixture, so the reader always ends. + self.reader.join(timeout=STOP_TIMEOUT) + return elapsed + + def kill(self): + """Kill the CLI's process group and every descendant seen earlier.""" + for pid in (self.process.pid, *self.descendants): + with contextlib.suppress(ProcessLookupError): + os.kill(pid, signal.SIGKILL) + with contextlib.suppress(ProcessLookupError): + os.killpg(self.process.pid, signal.SIGKILL) + self.process.wait(timeout=STOP_TIMEOUT) + + def survivors(self) -> list[str]: + """Describe descendants still alive a moment after the CLI exited. + + Returns: + The surviving processes. + """ + deadline = time.monotonic() + 5 + while True: + alive = [pid for pid in self.descendants if _alive(pid)] + if not alive or time.monotonic() > deadline: + return [_describe(pid) for pid in alive] + time.sleep(0.1) + + def errors(self) -> list[str]: + """List output lines that report an error. + + Returns: + The offending lines, decoded. + """ + lines = self.output.splitlines() + if self.json_mode: + offending = [] + for line in lines: + try: + record = json.loads(line) + except ValueError: + offending.append(f"not JSON: {line}") + continue + if record.get("level") in ("error", "critical"): + offending.append(line) + return offending + return [ + line + for line in lines + if "failed with exit code" in line or "[ERROR]" in line + ] + + +@pytest.fixture +def running_app( + request: pytest.FixtureRequest, app_dir: Path +) -> Generator[_RunningApp, None, None]: + """Start the app for one case and make sure nothing outlives the test. + + Args: + request: The fixture request, parametrized with (json_mode, npm). + app_dir: The initialized app directory. + + Yields: + The running app. + """ + json_mode, npm = request.param + if npm and shutil.which("npm") is None: + pytest.skip("npm is not installed") + app = _RunningApp(app_dir, json_mode=json_mode, npm=npm) + try: + app.wait_until_running() + yield app + finally: + if app.process.poll() is None: + app.kill() + for pid in app.descendants: + with contextlib.suppress(ProcessLookupError): + os.kill(pid, signal.SIGKILL) + + +def _group_sigint(process: subprocess.Popen): + """Deliver a terminal Ctrl-C: SIGINT to the whole process group.""" + os.killpg(process.pid, signal.SIGINT) + + +def _pid_sigint(process: subprocess.Popen): + """Deliver SIGINT to the CLI's PID only, as ``kill -INT`` does.""" + process.send_signal(signal.SIGINT) + + +def _pid_sigterm(process: subprocess.Popen): + """Deliver SIGTERM to the CLI's PID only, as docker and systemd do.""" + process.send_signal(signal.SIGTERM) + + +@pytest.mark.parametrize( + "running_app", + [ + pytest.param((False, False), id="plain"), + pytest.param((True, False), id="json"), + pytest.param((False, True), id="plain-npm"), + ], + indirect=True, +) +@pytest.mark.parametrize( + "deliver", + [ + pytest.param(_group_sigint, id="group-sigint"), + pytest.param(_pid_sigint, id="pid-sigint"), + pytest.param(_pid_sigterm, id="pid-sigterm"), + ], +) +def test_run_stops_cleanly(running_app: _RunningApp, deliver): + """Every stop route ends the whole tree promptly without errors.""" + elapsed = running_app.stop(deliver) + + assert running_app.process.returncode in (0, 130, 143, -signal.SIGINT) + assert elapsed < STOP_TIMEOUT + assert running_app.survivors() == [] + assert running_app.errors() == [] diff --git a/tests/units/reflex_base/utils/test_log.py b/tests/units/reflex_base/utils/test_log.py index dfdf620957e..43bbaed5e65 100644 --- a/tests/units/reflex_base/utils/test_log.py +++ b/tests/units/reflex_base/utils/test_log.py @@ -1,13 +1,10 @@ """Tests for the standard logging pipeline in reflex_base.utils.log.""" -import contextlib import json import logging import os -import signal import subprocess import sys -import time from pathlib import Path from unittest import mock @@ -894,389 +891,3 @@ def test_reset_releases_the_reservation(): log.reserve_stdout() log._reset() assert log.is_stdout_reserved() is False - - -_SUPERVISED_SCRIPT = """ -import logging -import multiprocessing -import os -import subprocess -import sys - -from reflex_base.utils import log - - -def crash(): - raise RuntimeError("boom") - - -if __name__ == "__main__": - log.enable_managed_logging() - if not log.is_output_supervised(): - sys.exit(log.supervise_output([sys.executable, __file__])) - print("from print") - os.write(1, b"from os.write\\n") - sys.stderr.write("from stderr\\n") - subprocess.run([sys.executable, "-c", "print('from child')"], check=True) - subprocess.run( - [sys.executable, "-c", "import json; print(json.dumps({'passed': 1}))"], - check=True, - ) - record = '{"level": "info", "logger": "child", "message": "passed"}' - subprocess.run([sys.executable, "-c", f"print({record!r})"], check=True) - sys.stderr.write("Traceback (most recent call last):\\n File \\"x.py\\"\\n") - sys.stderr.write('{"level": "error", "logger": "other", "message": "mid"}\\n') - sys.stderr.write("KeyError: 'interleaved'\\n") - logging.getLogger("reflex").info("from logger") - worker = multiprocessing.get_context("spawn").Process(target=crash) - worker.start() - worker.join() - print("\\u2713 unicode") - print("partial line", end="") - sys.exit(3) -""" - - -def _run_script(tmp_path: Path, source: str, **kwargs) -> subprocess.CompletedProcess: - """Run a Python script in JSON mode. - - Args: - tmp_path: The directory for the script. - source: The script source. - kwargs: Extra arguments for subprocess.run. - - Returns: - The completed process, with bytes output. - """ - script = tmp_path / "script.py" - script.write_text(source) - env = {**kwargs.pop("env", os.environ), "REFLEX_LOG_JSON": "true"} - return subprocess.run( - [sys.executable, str(script)], - capture_output=True, - env=env, - cwd=tmp_path, - timeout=60, - **kwargs, - ) - - -def test_supervise_output_turns_every_line_into_json(tmp_path): - """Everything the command and its descendants print reaches the streams as JSON.""" - result = _run_script(tmp_path, _SUPERVISED_SCRIPT) - - assert result.returncode == 3, result.stderr - out = [json.loads(line) for line in result.stdout.splitlines()] - err = [json.loads(line) for line in result.stderr.splitlines()] - assert sorted(r["message"] for r in out if r.get("logger") == "stdout") == [ - "from child", - "from os.write", - "from print", - "partial line", - # JSON that is not a log record is wrapped like any other line. - '{"passed": 1}', - "\u2713 unicode", - ] - assert {"level": "info", "logger": "child", "message": "passed"} in out - assert {"level": "error", "logger": "other", "message": "mid"} in err - assert [r["message"] for r in out if r.get("logger") == "reflex"] == ["from logger"] - assert { - "logger": "stderr", - "level": "warning", - "message": "from stderr", - }.items() <= err[0].items() - interleaved, crash = [r for r in err if "exception" in r] - assert interleaved["message"] == "KeyError: 'interleaved'" - assert interleaved["exception"] == ( - "Traceback (most recent call last):\n File \"x.py\"\nKeyError: 'interleaved'\n" - ) - assert crash["level"] == "error" - assert crash["message"] == "RuntimeError: boom" - assert crash["exception"].startswith("Traceback (most recent call last):") - assert 'raise RuntimeError("boom")' in crash["exception"] - - -_BURST_SCRIPT = """ -import sys - -from reflex_base.utils import log - -if not log.is_output_supervised(): - sys.exit(log.supervise_output([sys.executable, __file__, *sys.argv[1:]])) -for i in range(200): - print(f"line {i:03d} " + "x" * 1000, file=getattr(sys, sys.argv[1])) -""" - - -@pytest.mark.parametrize("drain_max_seconds", [log._DRAIN_MAX_SECONDS, 0.05]) -@pytest.mark.parametrize("stream", ["stdout", "stderr"]) -def test_supervise_output_drains_everything_for_a_slow_consumer( - tmp_path, drain_max_seconds, stream -): - """Output still in the pipes at exit reaches a consumer that reads late.""" - script = tmp_path / "burst.py" - script.write_text( - f"from reflex_base.utils import log\n" - f"log._DRAIN_MAX_SECONDS = {drain_max_seconds}\n" + _BURST_SCRIPT - ) - with subprocess.Popen( - [sys.executable, str(script), stream], - stdout=subprocess.PIPE if stream == "stdout" else None, - stderr=subprocess.PIPE if stream == "stderr" else None, - env={**os.environ, "REFLEX_LOG_JSON": "true"}, - ) as proc: - output = proc.stdout if stream == "stdout" else proc.stderr - assert output is not None - lines = [] - # A consumer slower than the writer keeps the pipes full at exit. - for line in output: - lines.append(line) - time.sleep(0.015) - proc.wait(timeout=30) - assert proc.returncode == 0 - assert len(lines) == 200 - assert json.loads(lines[-1])["message"].startswith("line 199 ") - - -def test_output_pump_write_snapshots_use_a_consistent_cutoff(monkeypatch): - """Snapshots measure a blocked or completed write at their own cutoff.""" - pump = log._OutputPump(0, 1, "info", "stdout") - snapshots = [] - - def write(fd, data): - """Capture a snapshot while the simulated write is in progress. - - Args: - fd: The output file descriptor. - data: The bytes to write. - - Returns: - The number of bytes written. - """ - snapshots.append(pump.writing_snapshot()) - return len(data) - - monkeypatch.setattr( - time, "monotonic", mock.Mock(side_effect=[10.0, 15.0, 20.0, 25.0]) - ) - monkeypatch.setattr(os, "write", write) - pump._write([b"line"], [], final=False) - assert snapshots == [(15.0, 5.0)] - assert pump.writing_snapshot() == (25.0, 10.0) - - -_STALLED_CONSUMER_SCRIPT = """ -import os -import subprocess -import sys -from pathlib import Path - -from reflex_base.utils import log - -log._DRAIN_WALL_SECONDS = 0.5 -if not log.is_output_supervised(): - sys.exit(log.supervise_output([sys.executable, __file__])) -writer = subprocess.Popen([ - sys.executable, - "-c", - "import os\\nwhile True:\\n os.write(int(os.environ['TEST_OUTPUT_FD']), (b'x' * 1023 + b'\\\\n') * 64)", -]) -Path("writer.pid").write_text(str(writer.pid)) -""" - - -@pytest.mark.parametrize("stream_fd", [1, 2]) -def test_supervise_output_bounds_shutdown_for_a_stalled_consumer(tmp_path, stream_fd): - """A lingering writer and an unread output pipe cannot prevent shutdown.""" - script = tmp_path / "stalled.py" - script.write_text(_STALLED_CONSUMER_SCRIPT) - with subprocess.Popen( - [sys.executable, str(script)], - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - cwd=tmp_path, - env={ - **os.environ, - "REFLEX_LOG_JSON": "true", - "TEST_OUTPUT_FD": str(stream_fd), - }, - ) as proc: - try: - assert proc.wait(timeout=5) == 0 - finally: - if proc.poll() is None: - proc.kill() - proc.wait(timeout=5) - pid_file = tmp_path / "writer.pid" - if pid_file.exists(): - with contextlib.suppress(OSError): - os.kill(int(pid_file.read_text()), signal.SIGTERM) - - -_LINGERING_SCRIPT = """ -import subprocess -import sys - -from reflex_base.utils import log - -if not log.is_output_supervised(): - sys.exit(log.supervise_output([sys.executable, __file__])) -sleeper = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"]) -print(sleeper.pid) -""" - - -def test_supervise_output_does_not_wait_for_lingering_descendants(tmp_path): - """A descendant that outlives the command and holds the pipe does not block exit.""" - start = time.monotonic() - result = _run_script(tmp_path, _LINGERING_SCRIPT) - elapsed = time.monotonic() - start - [record] = [json.loads(line) for line in result.stdout.splitlines()] - with contextlib.suppress(OSError): - os.kill(int(record["message"]), signal.SIGTERM) - assert result.returncode == 0, result.stderr - assert elapsed < 10 - - -_SLEEP_SCRIPT = """ -import sys -import time - -from reflex_base.utils import log - -if not log.is_output_supervised(): - sys.exit(log.supervise_output([sys.executable, __file__])) -print("ready") -time.sleep(30) -""" - - -@pytest.mark.skipif( - sys.platform == "win32", reason="PID-targeted signals are POSIX-only" -) -@pytest.mark.parametrize( - "signum", [signal.SIGTERM, signal.SIGINT], ids=["sigterm", "sigint"] -) -def test_supervise_output_forwards_shutdown_signal(tmp_path, signum): - """A signal sent to the supervisor's PID stops the command and is reported.""" - script = tmp_path / "sleep.py" - script.write_text(_SLEEP_SCRIPT) - with subprocess.Popen( - [sys.executable, str(script)], - stdout=subprocess.PIPE, - env={**os.environ, "REFLEX_LOG_JSON": "true"}, - ) as proc: - assert proc.stdout is not None - assert json.loads(proc.stdout.readline())["message"] == "ready" - proc.send_signal(signum) - assert proc.wait(timeout=10) == 128 + signum - - -def test_output_pump_survives_a_gone_consumer(): - """The reader keeps draining the child when its consumer closed the stream.""" - child_read, child_write = os.pipe() - out_read, out_write = os.pipe() - os.close(out_read) - os.write(child_write, b"plain\nTraceback (most recent call last):\n File 'x'\n") - os.close(child_write) - pump = log._OutputPump(child_read, out_write, "info", "stdout") - pump.run() - os.close(out_write) - - -_ENCODING_SCRIPT = """ -import os -import sys - -from reflex_base.utils import log - -if not log.is_output_supervised(): - sys.exit(log.supervise_output([sys.executable, __file__])) -print("gr\\u00fc\\u00dfe") -sys.stdout.flush() -os.write(1, b"10%\\r50%\\r100%\\r\\n") -""" - - -def test_supervise_output_decodes_utf8_and_splits_progress(tmp_path): - """Non-ASCII survives any PYTHONIOENCODING, and carriage-return updates are lines.""" - result = _run_script( - tmp_path, _ENCODING_SCRIPT, env={**os.environ, "PYTHONIOENCODING": "latin-1"} - ) - assert result.returncode == 0, result.stderr - assert [json.loads(line)["message"] for line in result.stdout.splitlines()] == [ - "grüße", - "10%", - "50%", - "100%", - ] - - -_CHATTY_LINGERING_SCRIPT = """ -import subprocess -import sys - -from reflex_base.utils import log - -if not log.is_output_supervised(): - sys.exit(log.supervise_output([sys.executable, __file__])) -chatter = subprocess.Popen([ - sys.executable, - "-c", - "import time\\nfor _ in range(600):\\n print('tick', flush=True)\\n time.sleep(0.05)", -]) -print(chatter.pid) -""" - - -def test_supervise_output_does_not_wait_for_a_chatty_descendant(tmp_path): - """A descendant that keeps writing after the command exits does not block exit.""" - start = time.monotonic() - result = _run_script(tmp_path, _CHATTY_LINGERING_SCRIPT) - elapsed = time.monotonic() - start - # The descendant can print before the command prints its PID. - [pid] = [ - int(record["message"]) - for record in map(json.loads, result.stdout.splitlines()) - if record["message"].isdigit() - ] - with contextlib.suppress(OSError): - os.kill(pid, signal.SIGTERM) - assert result.returncode == 0, result.stderr - assert elapsed < 15 - - -_CHILD_ENV_SCRIPT = """ -import os -import sys - -from reflex_base.utils import log - -if not log.is_output_supervised(): - # Not os.getppid() in the child: a Windows venv python.exe is a launcher - # that runs the real interpreter as its own child. - os.environ["TEST_SUPERVISOR_PID"] = str(os.getpid()) - sys.exit(log.supervise_output([sys.executable, __file__])) -print(os.environ[log._SUPERVISED_ENV_VAR] == os.environ["TEST_SUPERVISOR_PID"]) -print(sys.stdout.write_through) -print(sys.stdout.encoding, sys.stdout.errors) -""" - - -def test_supervise_output_child_environment(tmp_path): - """The child gets the supervisor PID, keeps user buffering and error handler, and writes UTF-8.""" - result = _run_script( - tmp_path, - _CHILD_ENV_SCRIPT, - env={ - **os.environ, - "PYTHONUNBUFFERED": "", - "PYTHONIOENCODING": "latin-1:backslashreplace", - }, - ) - assert result.returncode == 0, result.stderr - assert [json.loads(line)["message"] for line in result.stdout.splitlines()] == [ - "True", - "False", - "utf-8 backslashreplace", - ] diff --git a/tests/units/reflex_base/utils/test_process_tree.py b/tests/units/reflex_base/utils/test_process_tree.py new file mode 100644 index 00000000000..5278e9a593f --- /dev/null +++ b/tests/units/reflex_base/utils/test_process_tree.py @@ -0,0 +1,237 @@ +"""Tests for stopping a process together with everything it started.""" + +from __future__ import annotations + +import contextlib +import os +import signal +import subprocess +import sys +import time +from pathlib import Path + +import psutil +import pytest +from reflex_base.utils import process_tree + +# A child that starts a grandchild and does not forward signals to it, like +# ``npm run`` in front of a dev server. It reports the grandchild PID on stdout. +_ORPHANING_CHILD = """ +import subprocess, sys, time +grandchild = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"]) +print(grandchild.pid, flush=True) +time.sleep(60) +""" + +# A child that ignores SIGTERM, so only SIGKILL stops it. +_STUBBORN_CHILD = """ +import signal, time +signal.signal(signal.SIGTERM, signal.SIG_IGN) +print("ready", flush=True) +time.sleep(60) +""" + + +def _alive(pid: int) -> bool: + """Check whether a process still exists. + + Args: + pid: The process ID. + + Returns: + Whether the process exists (a zombie counts as gone). + """ + if sys.platform == "win32": + try: + return psutil.Process(pid).status() != psutil.STATUS_ZOMBIE + except psutil.NoSuchProcess: + return False + try: + os.kill(pid, 0) + if sys.platform == "linux": + stat = Path(f"/proc/{pid}/stat").read_bytes() + return stat.rsplit(b")", 1)[1].split()[0] != b"Z" + except OSError: + return False + return True + + +def _wait_gone(pid: int, timeout: float = 5) -> bool: + """Wait until a process is gone. + + Args: + pid: The process ID. + timeout: How long to wait. + + Returns: + Whether the process went away in time. + """ + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if not _alive(pid): + return True + time.sleep(0.02) + return not _alive(pid) + + +def _start(tmp_path: Path, source: str) -> tuple[subprocess.Popen, str]: + """Start a scripted child and read its first output line. + + Args: + tmp_path: The directory for the script. + source: The script source. + + Returns: + The process and its first output line. + """ + script = tmp_path / "child.py" + script.write_text(source) + proc = subprocess.Popen( + [sys.executable, str(script)], stdout=subprocess.PIPE, text=True + ) + assert proc.stdout is not None + return proc, proc.stdout.readline().strip() + + +def test_descendant_pids_lists_the_whole_tree(tmp_path: Path): + """Every level below the process is found, parents before children.""" + proc, line = _start(tmp_path, _ORPHANING_CHILD) + grandchild = int(line) + try: + if sys.platform == "win32": + pytest.skip("taskkill walks the tree on Windows") + assert process_tree.descendant_pids(proc.pid) == [grandchild] + tree = process_tree.descendant_pids(os.getpid()) + assert tree.index(proc.pid) < tree.index(grandchild) + finally: + process_tree.stop_process_tree(proc, timeout=1) + + +@pytest.mark.skipif(sys.platform != "linux", reason="/proc is Linux-only") +def test_descendant_pids_falls_back_to_ps(tmp_path: Path, monkeypatch): + """Without /proc the tree is read from ps, as on macOS.""" + monkeypatch.setattr(process_tree, "_PROC", tmp_path / "no-proc") + proc, line = _start(tmp_path, _ORPHANING_CHILD) + try: + assert process_tree.descendant_pids(proc.pid) == [int(line)] + finally: + process_tree.stop_process_tree(proc, timeout=1) + + +def test_stop_process_tree_stops_an_orphaned_grandchild(tmp_path: Path): + """A descendant the child does not stop itself is stopped too.""" + proc, line = _start(tmp_path, _ORPHANING_CHILD) + grandchild = int(line) + try: + started = time.monotonic() + process_tree.stop_process_tree(proc, timeout=5) + assert time.monotonic() - started < 4 + assert proc.returncode is not None + assert _wait_gone(grandchild) + finally: + # SIGTERM exists on every platform; on Windows it terminates outright. + with contextlib.suppress(OSError): + os.kill(grandchild, signal.SIGTERM) + + +# The killed grandchild stays a zombie under its living parent, which only +# the Linux check sees as gone. +@pytest.mark.skipif(sys.platform != "linux", reason="zombie detection needs /proc") +def test_kill_descendants_kills_below_the_root_only(tmp_path: Path): + """Every descendant is killed at once and the root keeps running.""" + proc, line = _start(tmp_path, _ORPHANING_CHILD) + grandchild = int(line) + try: + process_tree.kill_descendants(proc.pid) + assert _wait_gone(grandchild, timeout=1) + assert proc.poll() is None + finally: + process_tree.stop_process_tree(proc, timeout=1) + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX signals") +def test_kill_descendants_kills_known_pids_no_longer_below_the_root(tmp_path: Path): + """A descendant seen earlier is killed although its parent has exited.""" + proc, line = _start(tmp_path, _ORPHANING_CHILD) + grandchild = int(line) + root = _sleeper() + try: + proc.terminate() + proc.wait(timeout=5) + assert grandchild not in process_tree.descendant_pids(root.pid) + process_tree.kill_descendants(root.pid, [grandchild]) + assert _wait_gone(grandchild, timeout=1) + finally: + root.kill() + root.wait() + with contextlib.suppress(OSError): + os.kill(grandchild, signal.SIGKILL) + + +def _sleeper() -> subprocess.Popen: + """Start a child with no children of its own, to serve as a root. + + Returns: + The child process. + """ + return subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"]) + + +# A child that exits at once, leaving a grandchild in a process group of its +# own: it stands in for an unrelated process that took over a reused ID. +_DETACHING_CHILD = """ +import subprocess, sys +grandchild = subprocess.Popen( + [sys.executable, "-c", "import time; time.sleep(60)"], start_new_session=True +) +print(grandchild.pid, flush=True) +""" + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX signals") +def test_kill_descendants_spares_a_known_pid_outside_the_group(tmp_path: Path): + """A known ID now held by a process outside the root's group is not killed.""" + proc, line = _start(tmp_path, _DETACHING_CHILD) + grandchild = int(line) + root = _sleeper() + try: + proc.wait(timeout=5) + process_tree.kill_descendants(root.pid, [grandchild]) + time.sleep(0.2) + assert _alive(grandchild) + finally: + root.kill() + root.wait() + with contextlib.suppress(OSError): + os.kill(grandchild, signal.SIGKILL) + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX signals") +def test_stop_process_tree_kills_a_child_that_ignores_sigterm(tmp_path: Path): + """A child that survives the deadline is killed.""" + proc, line = _start(tmp_path, _STUBBORN_CHILD) + assert line == "ready" + started = time.monotonic() + process_tree.stop_process_tree(proc, timeout=0.5) + assert proc.returncode == -signal.SIGKILL + assert 0.5 <= time.monotonic() - started < 3 + + +def test_stop_process_tree_on_an_exited_process(tmp_path: Path): + """Stopping a process that already exited only reaps it.""" + proc = subprocess.Popen([sys.executable, "-c", "pass"]) + proc.wait() + process_tree.stop_process_tree(proc, timeout=1) + assert proc.returncode == 0 + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX process states") +def test_alive_treats_an_unreaped_child_as_gone(): + """A zombie no longer runs, so waiting for it would only waste the timeout.""" + proc = subprocess.Popen([sys.executable, "-c", "pass"]) + deadline = time.monotonic() + 5 + while process_tree._alive(proc.pid) and time.monotonic() < deadline: + time.sleep(0.02) + assert not process_tree._alive(proc.pid) + # _alive reaped the zombie, which is what happens to orphans under PID 1. + assert proc.poll() in (0, None) diff --git a/tests/units/reflex_base/utils/test_supervisor.py b/tests/units/reflex_base/utils/test_supervisor.py new file mode 100644 index 00000000000..c86447a17e7 --- /dev/null +++ b/tests/units/reflex_base/utils/test_supervisor.py @@ -0,0 +1,641 @@ +"""Tests for the ``reflex run --json`` output supervisor.""" + +import contextlib +import json +import os +import signal +import subprocess +import sys +import threading +import time +from pathlib import Path +from unittest import mock + +import pytest +from reflex_base.utils import supervisor + +_SUPERVISED_SCRIPT = """ +import logging +import multiprocessing +import os +import subprocess +import sys + +from reflex_base.utils import log, supervisor + + +def crash(): + raise RuntimeError("boom") + + +if __name__ == "__main__": + log.enable_managed_logging() + if not supervisor.is_output_supervised(): + sys.exit(supervisor.supervise_output([sys.executable, __file__])) + print("from print") + os.write(1, b"from os.write\\n") + sys.stderr.write("from stderr\\n") + subprocess.run([sys.executable, "-c", "print('from child')"], check=True) + subprocess.run( + [sys.executable, "-c", "import json; print(json.dumps({'passed': 1}))"], + check=True, + ) + record = '{"level": "info", "logger": "child", "message": "passed"}' + subprocess.run([sys.executable, "-c", f"print({record!r})"], check=True) + sys.stderr.write("Traceback (most recent call last):\\n File \\"x.py\\"\\n") + sys.stderr.write('{"level": "error", "logger": "other", "message": "mid"}\\n') + sys.stderr.write("KeyError: 'interleaved'\\n") + logging.getLogger("reflex").info("from logger") + worker = multiprocessing.get_context("spawn").Process(target=crash) + worker.start() + worker.join() + print("\\u2713 unicode") + print("partial line", end="") + sys.exit(3) +""" + + +def _run_script(tmp_path: Path, source: str, **kwargs) -> subprocess.CompletedProcess: + """Run a Python script in JSON mode. + + Args: + tmp_path: The directory for the script. + source: The script source. + kwargs: Extra arguments for subprocess.run. + + Returns: + The completed process, with bytes output. + """ + script = tmp_path / "script.py" + script.write_text(source) + env = {**kwargs.pop("env", os.environ), "REFLEX_LOG_JSON": "true"} + return subprocess.run( + [sys.executable, str(script)], + capture_output=True, + env=env, + cwd=tmp_path, + timeout=60, + **kwargs, + ) + + +def test_supervise_output_turns_every_line_into_json(tmp_path): + """Everything the command and its descendants print reaches the streams as JSON.""" + result = _run_script(tmp_path, _SUPERVISED_SCRIPT) + + assert result.returncode == 3, result.stderr + out = [json.loads(line) for line in result.stdout.splitlines()] + err = [json.loads(line) for line in result.stderr.splitlines()] + assert sorted(r["message"] for r in out if r.get("logger") == "stdout") == [ + "from child", + "from os.write", + "from print", + "partial line", + # JSON that is not a log record is wrapped like any other line. + '{"passed": 1}', + "\u2713 unicode", + ] + assert {"level": "info", "logger": "child", "message": "passed"} in out + assert {"level": "error", "logger": "other", "message": "mid"} in err + assert [r["message"] for r in out if r.get("logger") == "reflex"] == ["from logger"] + assert { + "logger": "stderr", + "level": "warning", + "message": "from stderr", + }.items() <= err[0].items() + interleaved, crash = [r for r in err if "exception" in r] + assert interleaved["message"] == "KeyError: 'interleaved'" + assert interleaved["exception"] == ( + "Traceback (most recent call last):\n File \"x.py\"\nKeyError: 'interleaved'\n" + ) + assert crash["level"] == "error" + assert crash["message"] == "RuntimeError: boom" + assert crash["exception"].startswith("Traceback (most recent call last):") + assert 'raise RuntimeError("boom")' in crash["exception"] + + +_BURST_SCRIPT = """ +import sys + +from reflex_base.utils import supervisor + +if not supervisor.is_output_supervised(): + sys.exit(supervisor.supervise_output([sys.executable, __file__, *sys.argv[1:]])) +for i in range(200): + print(f"line {i:03d} " + "x" * 1000, file=getattr(sys, sys.argv[1])) +""" + + +@pytest.mark.parametrize("drain_max_seconds", [supervisor._DRAIN_MAX_SECONDS, 0.05]) +@pytest.mark.parametrize("stream", ["stdout", "stderr"]) +def test_supervise_output_drains_everything_for_a_slow_consumer( + tmp_path, drain_max_seconds, stream +): + """Output still in the pipes at exit reaches a consumer that reads late.""" + script = tmp_path / "burst.py" + script.write_text( + f"from reflex_base.utils import supervisor\n" + f"supervisor._DRAIN_MAX_SECONDS = {drain_max_seconds}\n" + _BURST_SCRIPT + ) + with subprocess.Popen( + [sys.executable, str(script), stream], + stdout=subprocess.PIPE if stream == "stdout" else None, + stderr=subprocess.PIPE if stream == "stderr" else None, + env={**os.environ, "REFLEX_LOG_JSON": "true"}, + ) as proc: + output = proc.stdout if stream == "stdout" else proc.stderr + assert output is not None + lines = [] + # A consumer slower than the writer keeps the pipes full at exit. + for line in output: + lines.append(line) + time.sleep(0.015) + proc.wait(timeout=30) + assert proc.returncode == 0 + assert len(lines) == 200 + assert json.loads(lines[-1])["message"].startswith("line 199 ") + + +def test_output_pump_write_snapshots_use_a_consistent_cutoff(monkeypatch): + """Snapshots measure a blocked or completed write at their own cutoff.""" + pump = supervisor._OutputPump(0, 1, "info", "stdout") + snapshots = [] + + def write(fd, data): + """Capture a snapshot while the simulated write is in progress. + + Args: + fd: The output file descriptor. + data: The bytes to write. + + Returns: + The number of bytes written. + """ + snapshots.append(pump.writing_snapshot()) + return len(data) + + monkeypatch.setattr( + time, "monotonic", mock.Mock(side_effect=[10.0, 15.0, 20.0, 25.0]) + ) + monkeypatch.setattr(os, "write", write) + pump._write([b"line"], [], final=False) + assert snapshots == [(15.0, 5.0)] + assert pump.writing_snapshot() == (25.0, 10.0) + + +_STALLED_CONSUMER_SCRIPT = """ +import os +import subprocess +import sys +from pathlib import Path + +from reflex_base.utils import supervisor + +supervisor._DRAIN_WALL_SECONDS = 0.5 +if not supervisor.is_output_supervised(): + sys.exit(supervisor.supervise_output([sys.executable, __file__])) +writer = subprocess.Popen([ + sys.executable, + "-c", + "import os\\nwhile True:\\n os.write(int(os.environ['TEST_OUTPUT_FD']), (b'x' * 1023 + b'\\\\n') * 64)", +]) +Path("writer.pid").write_text(str(writer.pid)) +""" + + +@pytest.mark.parametrize("stream_fd", [1, 2]) +def test_supervise_output_bounds_shutdown_for_a_stalled_consumer(tmp_path, stream_fd): + """A lingering writer and an unread output pipe cannot prevent shutdown.""" + script = tmp_path / "stalled.py" + script.write_text(_STALLED_CONSUMER_SCRIPT) + with subprocess.Popen( + [sys.executable, str(script)], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + cwd=tmp_path, + env={ + **os.environ, + "REFLEX_LOG_JSON": "true", + "TEST_OUTPUT_FD": str(stream_fd), + }, + ) as proc: + try: + assert proc.wait(timeout=5) == 0 + finally: + if proc.poll() is None: + proc.kill() + proc.wait(timeout=5) + pid_file = tmp_path / "writer.pid" + if pid_file.exists(): + with contextlib.suppress(OSError): + os.kill(int(pid_file.read_text()), signal.SIGTERM) + + +_LINGERING_SCRIPT = """ +import subprocess +import sys + +from reflex_base.utils import supervisor + +if not supervisor.is_output_supervised(): + sys.exit(supervisor.supervise_output([sys.executable, __file__])) +sleeper = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"]) +print(sleeper.pid) +""" + + +def test_supervise_output_does_not_wait_for_lingering_descendants(tmp_path): + """A descendant that outlives the command and holds the pipe does not block exit.""" + start = time.monotonic() + result = _run_script(tmp_path, _LINGERING_SCRIPT) + elapsed = time.monotonic() - start + [record] = [json.loads(line) for line in result.stdout.splitlines()] + with contextlib.suppress(OSError): + os.kill(int(record["message"]), signal.SIGTERM) + assert result.returncode == 0, result.stderr + assert elapsed < 10 + + +_SLEEP_SCRIPT = """ +import sys +import time + +from reflex_base.utils import supervisor + +if not supervisor.is_output_supervised(): + sys.exit(supervisor.supervise_output([sys.executable, __file__])) +print("ready") +time.sleep(30) +""" + + +@pytest.mark.skipif( + sys.platform == "win32", reason="PID-targeted signals are POSIX-only" +) +@pytest.mark.parametrize( + "signum", [signal.SIGTERM, signal.SIGINT], ids=["sigterm", "sigint"] +) +def test_supervise_output_forwards_shutdown_signal(tmp_path, signum): + """A signal sent to the supervisor's PID stops the command and is reported.""" + script = tmp_path / "sleep.py" + script.write_text(_SLEEP_SCRIPT) + with subprocess.Popen( + [sys.executable, str(script)], + stdout=subprocess.PIPE, + env={**os.environ, "REFLEX_LOG_JSON": "true"}, + ) as proc: + assert proc.stdout is not None + assert json.loads(proc.stdout.readline())["message"] == "ready" + proc.send_signal(signum) + assert proc.wait(timeout=10) == 128 + signum + + +def _kill_supervised(proc: subprocess.Popen, tmp_path: Path): + """Kill and reap a test supervisor and the child it recorded in child.pid. + + Args: + proc: The supervisor process. + tmp_path: The directory holding the child's PID file. + """ + if proc.poll() is None: + proc.kill() + proc.wait(timeout=5) + pid_file = tmp_path / "child.pid" + if pid_file.exists(): + with contextlib.suppress(OSError, ValueError): + os.kill(int(pid_file.read_text()), signal.SIGKILL) + + +_CHATTY_CLEANUP_SCRIPT = """ +import os +import signal +import sys +import time +from pathlib import Path + +from reflex_base.utils import supervisor + +supervisor._STOP_DEADLINE_SECONDS = 1 +supervisor._DRAIN_WALL_SECONDS = 1 +if not supervisor.is_output_supervised(): + sys.exit(supervisor.supervise_output([sys.executable, __file__])) +Path("child.pid").write_text(str(os.getpid())) +def cleanup(*_): + for i in range(2000): + print(f"cleanup {i:05d} " + "x" * 200, flush=True) + raise SystemExit(0) +signal.signal(signal.SIGINT, cleanup) +print("ready", flush=True) +while True: + time.sleep(0.05) +""" + + +@pytest.mark.skipif(sys.platform == "win32", reason="PID-targeted signals") +def test_supervise_output_bounds_shutdown_when_the_consumer_stops_reading(tmp_path): + """A command blocked on a full pipe during cleanup is killed at the deadline.""" + script = tmp_path / "chatty_cleanup.py" + script.write_text(_CHATTY_CLEANUP_SCRIPT) + with subprocess.Popen( + [sys.executable, str(script)], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + cwd=tmp_path, + env={**os.environ, "REFLEX_LOG_JSON": "true"}, + ) as proc: + try: + assert proc.stdout is not None + assert json.loads(proc.stdout.readline())["message"] == "ready" + # Nothing reads stdout from here on. + started = time.monotonic() + proc.send_signal(signal.SIGINT) + # The deadline sends SIGTERM to the tree first, then SIGKILL. + assert proc.wait(timeout=15) in (128 + signal.SIGTERM, 128 + signal.SIGKILL) + assert time.monotonic() - started < 8 + finally: + _kill_supervised(proc, tmp_path) + + +_SECOND_INTERRUPT_SCRIPT = """ +import signal +import sys +import time + +from reflex_base.utils import supervisor + +if not supervisor.is_output_supervised(): + sys.exit(supervisor.supervise_output([sys.executable, __file__])) +count = 0 +def interrupted(*_): + global count + count += 1 + print(f"SIGINT #{count}", flush=True) + if count == 2: + raise SystemExit(0) +signal.signal(signal.SIGINT, interrupted) +print("ready", flush=True) +while True: + time.sleep(0.05) +""" + + +@pytest.mark.skipif(sys.platform == "win32", reason="PID-targeted signals") +def test_supervise_output_forwards_every_shutdown_signal(tmp_path): + """A repeated signal reaches the command; it decides what a repeat means.""" + script = tmp_path / "second_interrupt.py" + script.write_text(_SECOND_INTERRUPT_SCRIPT) + with subprocess.Popen( + [sys.executable, str(script)], + stdout=subprocess.PIPE, + env={**os.environ, "REFLEX_LOG_JSON": "true"}, + ) as proc: + try: + assert proc.stdout is not None + assert json.loads(proc.stdout.readline())["message"] == "ready" + proc.send_signal(signal.SIGINT) + time.sleep(0.3) + proc.send_signal(signal.SIGINT) + assert proc.wait(timeout=10) == 0 + messages = [json.loads(line)["message"] for line in proc.stdout] + assert messages == ["SIGINT #1", "SIGINT #2"] + finally: + _kill_supervised(proc, tmp_path) + + +_DEAF_SCRIPT = """ +import os +import signal +import sys +import time +from pathlib import Path + +from reflex_base.utils import supervisor + +supervisor._STOP_DEADLINE_SECONDS = 0.5 +if not supervisor.is_output_supervised(): + sys.exit(supervisor.supervise_output([sys.executable, __file__])) +Path("child.pid").write_text(str(os.getpid())) +signal.signal(signal.SIGINT, signal.SIG_IGN) +signal.signal(signal.SIGTERM, signal.SIG_IGN) +print("ready", flush=True) +time.sleep(30) +""" + + +@pytest.mark.skipif(sys.platform == "win32", reason="PID-targeted signals") +def test_supervise_output_kills_a_command_that_ignores_shutdown(tmp_path): + """A command that never stops on its own is killed when the deadline passes.""" + script = tmp_path / "deaf.py" + script.write_text(_DEAF_SCRIPT) + with subprocess.Popen( + [sys.executable, str(script)], + stdout=subprocess.PIPE, + cwd=tmp_path, + env={**os.environ, "REFLEX_LOG_JSON": "true"}, + ) as proc: + try: + assert proc.stdout is not None + assert json.loads(proc.stdout.readline())["message"] == "ready" + started = time.monotonic() + proc.send_signal(signal.SIGTERM) + assert proc.wait(timeout=10) == 128 + signal.SIGKILL + assert 0.5 <= time.monotonic() - started < 5 + finally: + _kill_supervised(proc, tmp_path) + + +_SLOW_CONSUMER_SCRIPT = """ +import sys + +from reflex_base.utils import supervisor + +supervisor._DRAIN_WALL_SECONDS = 0.5 +if not supervisor.is_output_supervised(): + sys.exit(supervisor.supervise_output([sys.executable, __file__])) +for i in range(300): + print(f"line {i:04d} " + "x" * 1000, flush=True) +""" + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX pipe semantics") +def test_supervise_output_drain_ends_between_records(tmp_path): + """When the drain gives up on a slow consumer, the stream ends at a record boundary.""" + script = tmp_path / "slow_consumer.py" + script.write_text(_SLOW_CONSUMER_SCRIPT) + with subprocess.Popen( + [sys.executable, str(script)], + stdout=subprocess.PIPE, + env={**os.environ, "REFLEX_LOG_JSON": "true"}, + ) as proc: + assert proc.stdout is not None + fd = proc.stdout.fileno() + chunks = [] + deadline = time.monotonic() + 20 + # Read about 40 KB/s: far slower than the writer, faster than the grace. + while time.monotonic() < deadline: + chunk = os.read(fd, 4096) + if not chunk: + break + chunks.append(chunk) + time.sleep(0.1) + assert proc.wait(timeout=10) == 0 + output = b"".join(chunks) + assert output.endswith(b"\n") + records = [json.loads(line) for line in output.splitlines()] + assert all(record["message"].startswith("line ") for record in records) + # The wall deadline cut the stream, so the boundary was actually enforced. + assert len(records) < 300 + + +def test_output_pump_survives_a_gone_consumer(): + """The reader keeps draining the child when its consumer closed the stream.""" + child_read, child_write = os.pipe() + out_read, out_write = os.pipe() + os.close(out_read) + os.write(child_write, b"plain\nTraceback (most recent call last):\n File 'x'\n") + os.close(child_write) + pump = supervisor._OutputPump(child_read, out_write, "info", "stdout") + pump.run() + os.close(out_write) + + +_ENCODING_SCRIPT = """ +import os +import sys + +from reflex_base.utils import supervisor + +if not supervisor.is_output_supervised(): + sys.exit(supervisor.supervise_output([sys.executable, __file__])) +print("gr\\u00fc\\u00dfe") +sys.stdout.flush() +os.write(1, b"10%\\r50%\\r100%\\r\\n") +""" + + +def test_supervise_output_decodes_utf8_and_splits_progress(tmp_path): + """Non-ASCII survives any PYTHONIOENCODING, and carriage-return updates are lines.""" + result = _run_script( + tmp_path, _ENCODING_SCRIPT, env={**os.environ, "PYTHONIOENCODING": "latin-1"} + ) + assert result.returncode == 0, result.stderr + assert [json.loads(line)["message"] for line in result.stdout.splitlines()] == [ + "grüße", + "10%", + "50%", + "100%", + ] + + +_CHATTY_LINGERING_SCRIPT = """ +import subprocess +import sys + +from reflex_base.utils import supervisor + +if not supervisor.is_output_supervised(): + sys.exit(supervisor.supervise_output([sys.executable, __file__])) +chatter = subprocess.Popen([ + sys.executable, + "-c", + "import time\\nfor _ in range(600):\\n print('tick', flush=True)\\n time.sleep(0.05)", +]) +print(chatter.pid) +""" + + +def test_supervise_output_does_not_wait_for_a_chatty_descendant(tmp_path): + """A descendant that keeps writing after the command exits does not block exit.""" + start = time.monotonic() + result = _run_script(tmp_path, _CHATTY_LINGERING_SCRIPT) + elapsed = time.monotonic() - start + # The descendant can print before the command prints its PID. + [pid] = [ + int(record["message"]) + for record in map(json.loads, result.stdout.splitlines()) + if record["message"].isdigit() + ] + with contextlib.suppress(OSError): + os.kill(pid, signal.SIGTERM) + assert result.returncode == 0, result.stderr + assert elapsed < 15 + + +_CHILD_ENV_SCRIPT = """ +import os +import sys + +from reflex_base.utils import supervisor + +if not supervisor.is_output_supervised(): + # Not os.getppid() in the child: a Windows venv python.exe is a launcher + # that runs the real interpreter as its own child. + os.environ["TEST_SUPERVISOR_PID"] = str(os.getpid()) + sys.exit(supervisor.supervise_output([sys.executable, __file__])) +print(os.environ[supervisor._SUPERVISED_ENV_VAR] == os.environ["TEST_SUPERVISOR_PID"]) +print(sys.stdout.write_through) +print(sys.stdout.encoding, sys.stdout.errors) +""" + + +def test_supervise_output_child_environment(tmp_path): + """The child gets the supervisor PID, keeps user buffering and error handler, and writes UTF-8.""" + result = _run_script( + tmp_path, + _CHILD_ENV_SCRIPT, + env={ + **os.environ, + "PYTHONUNBUFFERED": "", + "PYTHONIOENCODING": "latin-1:backslashreplace", + }, + ) + assert result.returncode == 0, result.stderr + assert [json.loads(line)["message"] for line in result.stdout.splitlines()] == [ + "True", + "False", + "utf-8 backslashreplace", + ] + + +def test_output_pump_writes_in_atomic_chunks(monkeypatch): + """Records are grouped into writes no larger than PIPE_BUF, one record minimum.""" + writes: list[bytes] = [] + monkeypatch.setattr(supervisor, "_write_all", lambda _fd, data: writes.append(data)) + pump = supervisor._OutputPump(0, 1, "info", "stdout") + lines = [b"x" * 100] * 200 + [b"y" * (2 * supervisor._PIPE_BUF)] + [b"z"] + + pump._write(lines, [], final=False) + + assert len(writes) > 3 + small, big, last = writes[:-2], writes[-2], writes[-1] + assert all(len(chunk) <= supervisor._PIPE_BUF for chunk in small) + assert all(chunk.endswith(b"\n") for chunk in writes) + assert json.loads(big)["message"] == "y" * (2 * supervisor._PIPE_BUF) + assert json.loads(last)["message"] == "z" + assert b"".join(writes).count(b"\n") == 202 + + +def test_output_pump_drain_grace_ends_with_the_write(monkeypatch): + """The grace ends when the write completes, even while the pipe stays open.""" + release = threading.Event() + started = threading.Event() + + def slow_write(_fd, _data): + started.set() + release.wait(5) + + monkeypatch.setattr(supervisor, "_write_all", slow_write) + # The write end stays open, like a descendant that outlives the command. + read_fd, write_fd = os.pipe() + pump = supervisor._OutputPump(read_fd, 1, "info", "stdout") + pump.start() + try: + os.write(write_fd, b"line\n") + assert started.wait(5) + threading.Timer(0.2, release.set).start() + begin = time.monotonic() + pump.drain(begin, 0.0, begin) + assert time.monotonic() - begin < 1 + finally: + os.close(write_fd) + pump.join(5) diff --git a/tests/units/test_reflex.py b/tests/units/test_reflex.py index e2923104a86..d1323442da6 100644 --- a/tests/units/test_reflex.py +++ b/tests/units/test_reflex.py @@ -5,6 +5,7 @@ import contextlib import json import os +import select import signal import subprocess import sys @@ -970,8 +971,13 @@ def record_process(*args,**kwargs): @pytest.mark.skipif(sys.platform == "win32", reason="POSIX signals") @pytest.mark.parametrize("mode", ["frontend", "fullstack"]) @pytest.mark.parametrize("sig", ["SIGTERM", "SIGINT"]) -def test_no_tty_run_stops_frontend(tmp_path, mode, sig): - """Headless runs stop the frontend on SIGTERM or SIGINT.""" +@pytest.mark.parametrize("forwards", [True, False], ids=["forwarding", "orphaning"]) +def test_no_tty_run_stops_frontend(tmp_path, mode, sig, forwards): + """Headless runs stop the frontend on SIGTERM or SIGINT. + + The frontend stand-in either passes SIGTERM on to its own child, as bun + does, or exits without it, as ``npm run`` does; the child must stop either way. + """ pids = tmp_path / "children" driver = tmp_path / "driver.py" driver.write_text( @@ -981,7 +987,7 @@ def test_no_tty_run_stops_frontend(tmp_path, mode, sig): def frontend(*args): code = "import signal,subprocess,sys,time\\n" + \ "p=subprocess.Popen([sys.executable,'-c','import time;time.sleep(60)'])\\n" + \ - "signal.signal(signal.SIGTERM, lambda *a: (p.terminate(), sys.exit(0)))\\n" + \ + ("signal.signal(signal.SIGTERM, lambda *a: (p.terminate(), sys.exit(0)))\\n" if FORWARDS else "") + \ "print(p.pid,flush=True);time.sleep(60)\\n" p = subprocess.Popen([sys.executable,"-c",code],stdout=subprocess.PIPE,text=True) grandchild = int(p.stdout.readline()) @@ -1008,6 +1014,7 @@ def stop(sig,frame): raise SystemExit(0) rx._run_dev(MODE,3000,PORT,"127.0.0.1") """ .replace("PIDS", repr(str(pids))) + .replace("FORWARDS", repr(forwards)) .replace( "MODE", "constants.RunningMode.FRONTEND_ONLY" @@ -1038,11 +1045,14 @@ def stop(sig,frame): raise SystemExit(0) time.sleep(0.01) assert child is not None, "frontend did not start" assert grandchild is not None, "frontend grandchild did not start" + started = time.monotonic() os.kill(launcher.pid, getattr(signal, sig)) try: returncode = launcher.wait(timeout=DEFAULT_TIMEOUT) except subprocess.TimeoutExpired: pytest.fail("no-TTY launcher hung after signal") + # Stopping the tree takes one SIGTERM round, well below the kill timeout. + assert time.monotonic() - started < 4 if sig == "SIGINT": assert returncode in (0, -signal.SIGINT) else: @@ -1074,6 +1084,213 @@ def stop(sig,frame): raise SystemExit(0) os.kill(pid, signal.SIGKILL) +_STUCK_CLEANUP_SCRIPT = """ +import os, signal, subprocess, sys, threading, time + +from reflex import reflex + +with reflex._shutdown_signals() as begin_cleanup: + try: + served = threading.Event() + if os.environ.get("SERVER_HANDLERS"): + # Like granian: its own handlers while serving, never restored. + for sig in (signal.SIGINT, signal.SIGTERM): + signal.signal(sig, lambda *_: served.set()) + print("ready", flush=True) + while not served.is_set(): + time.sleep(0.05) + finally: + begin_cleanup() + # A cleanup that never ends, like a frontend pipe held open forever. + helper = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"]) + print(helper.pid, flush=True) + while True: + time.sleep(1) +""" + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX signals") +@pytest.mark.parametrize("sig", [signal.SIGINT, signal.SIGTERM]) +@pytest.mark.parametrize("server_handlers", [False, True], ids=["own", "server"]) +def test_shutdown_signals_force_exit_on_a_later_repeat(tmp_path, sig, server_handlers): + """A repeat right after the first signal is ignored; a later one forces the exit. + + A backend server that installed its own handlers and never restored them, + as granian does, cannot swallow the later repeat. + """ + script = tmp_path / "stuck.py" + script.write_text(_STUCK_CLEANUP_SCRIPT) + proc = subprocess.Popen( + [sys.executable, str(script)], + stdout=subprocess.PIPE, + bufsize=0, + env={**os.environ, "SERVER_HANDLERS": "1" if server_handlers else ""}, + ) + helper = None + try: + assert _read_line(proc) == "ready\n" + proc.send_signal(sig) + helper = int(_read_line(proc)) + # The same Ctrl-C delivered twice under --json: not a request to force. + proc.send_signal(sig) + time.sleep(0.3) + assert proc.poll() is None + time.sleep(reflex._REPEAT_IGNORE_SECONDS) + proc.send_signal(sig) + assert proc.wait(timeout=DEFAULT_TIMEOUT) == 128 + sig + _assert_gone(helper, "the forced exit left the cleanup's child running") + finally: + proc.kill() + proc.wait() + if helper is not None: + with contextlib.suppress(ProcessLookupError): + os.kill(helper, signal.SIGKILL) + + +def _read_line(proc: subprocess.Popen) -> str: + """Read one line from a child's unbuffered stdout, failing instead of hanging. + + Args: + proc: The child, started with ``stdout=PIPE`` and ``bufsize=0`` so no + later line is read ahead into a buffer ``select`` cannot see. + + Returns: + The decoded line, newline included. + """ + assert proc.stdout is not None + if not select.select([proc.stdout], [], [], DEFAULT_TIMEOUT)[0]: + pytest.fail("the child printed nothing in time") + return proc.stdout.readline().decode() + + +def _assert_gone(pid: int, message: str) -> None: + """Wait for a process to exit; an exited process nobody reaped counts. + + Args: + pid: The process ID. + message: The failure message when it keeps running. + """ + deadline = time.monotonic() + DEFAULT_TIMEOUT + while time.monotonic() < deadline: + try: + if psutil.Process(pid).status() == psutil.STATUS_ZOMBIE: + return + except psutil.NoSuchProcess: + return + time.sleep(0.05) + pytest.fail(message) + + +@pytest.mark.skipif(sys.platform != "linux", reason="fills a pipe by whole pages") +def test_shutdown_signals_force_exit_with_a_full_stderr(tmp_path): + """A stderr nobody reads cannot hold up the forced exit.""" + script = tmp_path / "stuck.py" + script.write_text(_STUCK_CLEANUP_SCRIPT) + read_end, write_end = os.pipe() + proc = subprocess.Popen( + [sys.executable, str(script)], + stdout=subprocess.PIPE, + stderr=write_end, + bufsize=0, + ) + helper = None + try: + assert _read_line(proc) == "ready\n" + proc.send_signal(signal.SIGTERM) + helper = int(_read_line(proc)) + # Fill the pipe; its write end stays open here, so nothing drains it. + while select.select([], [write_end], [], 0)[1]: + os.write(write_end, b"\0" * select.PIPE_BUF) + time.sleep(reflex._REPEAT_IGNORE_SECONDS + 0.1) + proc.send_signal(signal.SIGTERM) + assert proc.wait(timeout=DEFAULT_TIMEOUT) == 128 + signal.SIGTERM + _assert_gone(helper, "the forced exit left the cleanup's child running") + finally: + proc.kill() + proc.wait() + os.close(read_end) + os.close(write_end) + if helper is not None: + with contextlib.suppress(ProcessLookupError): + os.kill(helper, signal.SIGKILL) + + +# A dev server that ignores SIGTERM behind a wrapper that exits on it, like a +# package manager in front of the server. +_SERVER_BEHIND_WRAPPER_SCRIPT = """ +import subprocess, sys, time + +server = subprocess.Popen( + [ + sys.executable, + "-c", + "import signal, time\\n" + "signal.signal(signal.SIGTERM, signal.SIG_IGN)\\n" + "print('ready', flush=True)\\n" + "time.sleep(60)", + ], + stdout=subprocess.PIPE, + text=True, +) +# Only report the server once it ignores SIGTERM. +server.stdout.readline() +print(server.pid, flush=True) +time.sleep(60) +""" + +_ORPHANING_CLEANUP_SCRIPT = """ +import subprocess, sys, time + +from reflex import reflex +from reflex_base.utils.process_tree import stop_process_tree + +wrapper = subprocess.Popen( + [sys.executable, sys.argv[1]], stdout=subprocess.PIPE, text=True +) +print(wrapper.stdout.readline(), end="", flush=True) +with reflex._shutdown_signals() as begin_cleanup: + try: + print("ready", flush=True) + while True: + time.sleep(1) + finally: + begin_cleanup() + # The wrapper exits at once; the server holds the stop until the deadline. + stop_process_tree(wrapper, timeout=60) +""" + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX signals") +def test_shutdown_signals_force_exit_kills_an_orphaned_descendant(tmp_path): + """The forced exit kills a server whose wrapper exited during the cleanup.""" + wrapper_script = tmp_path / "wrapper.py" + wrapper_script.write_text(_SERVER_BEHIND_WRAPPER_SCRIPT) + script = tmp_path / "stuck.py" + script.write_text(_ORPHANING_CLEANUP_SCRIPT) + proc = subprocess.Popen( + [sys.executable, str(script), str(wrapper_script)], + stdout=subprocess.PIPE, + bufsize=0, + ) + server = None + try: + server = int(_read_line(proc)) + assert _read_line(proc) == "ready\n" + proc.send_signal(signal.SIGTERM) + # By now the wrapper is gone and the server is no longer found below the run. + time.sleep(reflex._REPEAT_IGNORE_SECONDS + 0.2) + assert proc.poll() is None + proc.send_signal(signal.SIGTERM) + assert proc.wait(timeout=DEFAULT_TIMEOUT) == 128 + signal.SIGTERM + _assert_gone(server, "the forced exit left the orphaned server running") + finally: + proc.kill() + proc.wait() + if server is not None: + with contextlib.suppress(ProcessLookupError): + os.kill(server, signal.SIGKILL) + + @pytest.mark.parametrize( ("argv", "supervised", "expected"), [(["--json"], False, True), (["--json"], True, False), ([], False, False)], @@ -1095,14 +1312,14 @@ def test_run_supervises_output_only_in_json_mode( expected: Whether the supervisor is expected to start. """ from reflex_base.environment import environment - from reflex_base.utils import log + from reflex_base.utils import log, supervisor # Registered so teardown restores the variables the CLI callbacks set. monkeypatch.setenv(log._MANAGED_ENV_VAR, "true") monkeypatch.setenv(environment.REFLEX_LOG_JSON.name, "false") - monkeypatch.setenv(log._SUPERVISED_ENV_VAR, "1234" if supervised else "") + monkeypatch.setenv(supervisor._SUPERVISED_ENV_VAR, "1234" if supervised else "") monkeypatch.setattr(sys, "argv", ["reflex", "run", *argv]) - supervise = mocker.patch.object(log, "supervise_output", return_value=7) + supervise = mocker.patch.object(supervisor, "supervise_output", return_value=7) run = mocker.patch.object(reflex, "_run") mocker.patch("reflex.utils.prerequisites.check_running_mode") diff --git a/tests/units/utils/test_processes.py b/tests/units/utils/test_processes.py index b4917b258bd..7ecdd60f6a4 100644 --- a/tests/units/utils/test_processes.py +++ b/tests/units/utils/test_processes.py @@ -425,6 +425,20 @@ def test_stream_logs_keeps_posix_sigterm_codes_as_failures_on_windows( ) +def test_stream_logs_accepts_any_exit_code_after_a_stop_request(caplog): + """A process the caller asked to stop is not a failure, whatever its code. + + Args: + caplog: Pytest log capture. + """ + process = _finished_process(1) + + with caplog.at_level(logging.ERROR): + list(stream_logs("Starting frontend", process, expected_exit=lambda: True)) + + assert not caplog.records + + def test_stream_logs_still_fails_on_a_real_error_exit(caplog): """The relaxed set must not swallow an actual non-zero exit.