Skip to content

Commit 2a2a4c1

Browse files
Give reflex run one owner for shutdown (#7650)
* Give reflex run one owner for shutdown `reflex run` had no single place that stops its process tree. Each layer reacted to whatever signal reached it and relied on the terminal's process group, Python's KeyboardInterrupt, or exit-code conventions to reach the others. Every delivery route that differed from a terminal Ctrl-C became its own bug: a pid-targeted SIGTERM left `npm run`'s node server behind and hung the CLI on its pipe (#7485), the JSON supervisor's drain cap started only after the child exited so a stalled consumer blocked exit forever and a slow one got a cut record (#7510), and a second Ctrl-C could interrupt the cleanup the first one started. The shutdown now converges on one path at each layer: - `reflex_base.utils.processes.stop_process_tree` signals a process and every descendant it can find (/proc on Linux, ps elsewhere, taskkill /T on Windows), waits for all of them against one deadline and kills what is left. Processes stay in the CLI's process group, so a terminal Ctrl-C, Ctrl-Z or a group kill still reach them directly. - `_run_dev` installs idempotent SIGINT/SIGTERM handlers around the run: the first signal unwinds into cleanup, repeats are ignored. Its cleanup stops the whole frontend tree with that helper, on every platform, and `stream_logs` no longer reports an exit the CLI asked for as a failure. - `supervise_output` forwards every SIGINT/SIGTERM to the command, arms a 15 s deadline from the first one after which the command tree is killed, anchors the 30 s drain wall at the request, and writes record by record while stopping so the drain ends between two complete records. Regression coverage: unit tests for the tree helper, the no-TTY run with a frontend that does not forward signals, the supervisor with a stalled consumer during chatty cleanup, a repeated signal, a command that ignores shutdown, and a slow consumer at the drain cut; plus an integration matrix that stops a real `reflex run` by group SIGINT, pid SIGINT and pid SIGTERM in plain, --json and npm configurations and checks exit time, survivors and error output. Closes #7485 Closes #7510 Claude-Session: https://claude.ai/code/session_012Ps5W65XWRr7LRi3c4DvQU * Write JSON output in pipe-atomic chunks and treat zombies as exited Review findings on #7650: - A batch blocked in the reader thread when shutdown started was still one write, so an abandoned write could cut a record. Records are now grouped into writes of at most PIPE_BUF bytes at every stage, which a pipe accepts whole or not at all; the per-record mode after a stop request goes away with it. Write counts and throughput for a chatty child match main. - A descendant that exited but was never reaped, as happens for orphans under a non-reaping PID 1, counted as alive and cost the whole stop timeout. _alive now reaps orphans handed to this process and reads the Linux process state, so a zombie counts as gone. Claude-Session: https://claude.ai/code/session_012Ps5W65XWRr7LRi3c4DvQU * Stop the whole tree on Windows and guard the run's cleanup window Review findings on #7650 and the Windows unit-test failure: - The JSON supervisor forwarded SIGTERM on Windows as TerminateProcess on the command alone, leaving its workers behind; it now stops the whole tree with taskkill at once, since Windows has no graceful signal. - The run's SIGINT/SIGTERM guard is installed on every platform, and the run marks the start of its cleanup so a first signal arriving during the frontend stop is ignored instead of aborting it. - Test helpers no longer race a process that exits between the liveness probe and the /proc read, the tree test checks ordering by membership, and the Windows cleanup uses SIGTERM, which exists there. Claude-Session: https://claude.ai/code/session_012Ps5W65XWRr7LRi3c4DvQU * Use a Windows-safe pipe write size in the JSON supervisor select.PIPE_BUF exists only on POSIX; reading it on Windows raised in the reader thread as soon as a batch held two records. The size is now a module constant with the Linux value as fallback, the test reads that constant, and the supervisor docstring scopes the record-boundary guarantee to records that fit one write. Claude-Session: https://claude.ai/code/session_012Ps5W65XWRr7LRi3c4DvQU * Keep the new test imports at module level The slow-consumer drain test no longer shrinks the pipe with fcntl: the default pipe already holds far less than the test writes, so the cut happens regardless, and the test now runs on every POSIX platform. psutil is imported at the top of the tree test module like elsewhere in tests. Claude-Session: https://claude.ai/code/session_012Ps5W65XWRr7LRi3c4DvQU * End the drain grace with the write and drop the dead psutil kill The grace after the drain deadline waited for the reader thread to exit, so a descendant that still held the pipe open cost the full two seconds even after the write finished. It now waits for the write in progress. exec.kill lost its only caller when the frontend tree stop replaced the Windows workaround; remove it. Parse /proc/<pid>/stat in one helper. * Floor reflex-base at 0.10.2.dev0 for the process-tree helper reflex imports reflex_base.utils.processes, which is unreleased. The floor was 0.10.1.dev0, which the published reflex-base 0.10.1 satisfies without that module, so check-min-deps resolved to it and failed. Claude-Session: https://claude.ai/code/session_012Ps5W65XWRr7LRi3c4DvQU * Move the output supervisor out of log.py The JSON supervisor shares no code with the logging pipeline, so it now lives in reflex_base.utils.supervisor with its tests. The process tree helper is renamed to reflex_base.utils.process_tree so it no longer shares a name with reflex.utils.processes. No logic changes. * Force the exit on a later repeat of the stop signal A repeat within a second of the first SIGINT or SIGTERM is still the same request delivered twice and is ignored. A later one means the cleanup is stuck, for example on a frontend pipe that a process outside the tree holds open: every descendant is killed and reflex run exits at once. * Make the forced exit kill the saved tree and never block on stderr A forced exit found descendants by a fresh scan below reflex run, so a server whose wrapper had already exited during the cleanup, the npm case this branch fixes, survived it and kept its port. The cleanup now records everything below the process when it begins, and the forced exit kills that set as well as whatever a fresh scan finds. The forced exit also wrote its message before killing anything, so a stderr pipe nobody read could hold it up. The message is written only when stderr accepts it without blocking, after the kill. Claude-Session: https://claude.ai/code/session_012Ps5W65XWRr7LRi3c4DvQU * Bound the forced-exit tests' reads and note the Windows limit The scripted children are read line by line with a deadline from an unbuffered pipe, so a child that never prints fails the test instead of hanging the suite. The test that keeps a killed grandchild as a zombie under its living parent runs on Linux only, where the zombie is seen as gone. The docstrings say that on Windows a forced exit ends only the CLI, since a console Ctrl-C already reached every process on the console. Claude-Session: https://claude.ai/code/session_012Ps5W65XWRr7LRi3c4DvQU * Restore the shutdown handlers when the cleanup begins granian installs its own SIGINT/SIGTERM handlers while it serves and never restores ours, so after the backend returned a later Ctrl-C only set granian's flag and the forced exit never ran in a real app. begin_cleanup now puts the handlers back. * Spare a reused process ID on a forced exit The forced exit kills the process IDs recorded when the cleanup began. Between then and a later repeat of the signal an ID can be freed and taken by an unrelated process. A recorded ID is now killed only while it is still in the root's process group, which every process reflex run starts stays in; one that left the group belongs to someone else. Claude-Session: https://claude.ai/code/session_012Ps5W65XWRr7LRi3c4DvQU * Give the kill_descendants tests a root of their own The two tests for recorded IDs used the test runner as the root, so the kill reached every live descendant of pytest. They now kill below a child started for the purpose. Claude-Session: https://claude.ai/code/session_012Ps5W65XWRr7LRi3c4DvQU
1 parent bac5d98 commit 2a2a4c1

15 files changed

Lines changed: 2242 additions & 719 deletions

File tree

‎news/+run-shutdown-owner.bugfix.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
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.
Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
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.

‎packages/reflex-base/src/reflex_base/utils/log.py‎

Lines changed: 0 additions & 274 deletions
Original file line numberDiff line numberDiff line change
@@ -26,12 +26,8 @@
2626
import json
2727
import logging
2828
import os
29-
import re
3029
import shutil
31-
import signal
32-
import subprocess
3330
import sys
34-
import threading
3531
import time
3632
from collections.abc import Generator
3733
from pathlib import Path
@@ -264,276 +260,6 @@ def _write_json(payload: dict, *, stderr: bool):
264260
stream.flush()
265261

266262

267-
# The supervisor PID, set for the child of supervise_output() and inherited
268-
# by its descendants.
269-
_SUPERVISED_ENV_VAR = "REFLEX_OUTPUT_SUPERVISED"
270-
271-
# How long a reader may sit idle after the child exits before the supervisor
272-
# stops waiting for descendants that still hold the pipe.
273-
_DRAIN_IDLE_SECONDS = 0.5
274-
275-
# Maximum drain time after the child exits, excluding time forwarding output.
276-
_DRAIN_MAX_SECONDS = 5
277-
278-
# Stop shutdown after 30 seconds even when a consumer blocks every write.
279-
_DRAIN_WALL_SECONDS = 30
280-
281-
# Line ends in child output; a lone ``\r`` ends a progress-bar update.
282-
_LINE_END = re.compile(rb"\r\n|\r|\n")
283-
284-
285-
def is_output_supervised() -> bool:
286-
"""Check whether this process runs under supervise_output().
287-
288-
Returns:
289-
True if a parent process turns this process's output into JSON records.
290-
"""
291-
return bool(os.environ.get(_SUPERVISED_ENV_VAR))
292-
293-
294-
def _is_json_record(line: bytes) -> bool:
295-
"""Check whether a line already is a JSON log record.
296-
297-
Args:
298-
line: The line to check.
299-
300-
Returns:
301-
True if the line is a JSON object with a level and a message.
302-
"""
303-
if not line.startswith(b"{"):
304-
return False
305-
try:
306-
record = json.loads(line)
307-
except ValueError:
308-
return False
309-
return isinstance(record, dict) and "level" in record and "message" in record
310-
311-
312-
def _to_records(
313-
lines: list[bytes], traceback: list[str], level: str, name: str, final: bool
314-
) -> bytes:
315-
"""Convert raw output lines to JSON lines.
316-
317-
Lines that already are JSON log records pass through unchanged. A Python
318-
traceback is collected, across calls, into one record with the traceback
319-
text in its ``exception`` field.
320-
321-
Args:
322-
lines: Complete output lines, without their newline.
323-
traceback: The lines of a traceback still being collected.
324-
level: The record level for plain lines.
325-
name: The logger name for the records.
326-
final: Whether the output ended, so an open traceback is emitted too.
327-
328-
Returns:
329-
The JSON lines, each ending with a newline.
330-
"""
331-
timestamp = datetime.datetime.now(tz=datetime.UTC).isoformat()
332-
out: list[bytes] = []
333-
334-
def record(message: str, level: str = level, exception: str | None = None):
335-
payload = {
336-
"timestamp": timestamp,
337-
"level": level,
338-
"logger": name,
339-
"message": message,
340-
}
341-
if exception is not None:
342-
payload["exception"] = exception
343-
out.append(json.dumps(payload).encode())
344-
345-
for raw in lines:
346-
if _is_json_record(raw):
347-
out.append(raw)
348-
continue
349-
line = raw.decode("utf-8", "replace")
350-
if traceback:
351-
traceback.append(line)
352-
# Frames are indented; the exception line is not.
353-
if line.strip() and not line[0].isspace():
354-
record(line.strip(), "error", "\n".join(traceback) + "\n")
355-
traceback.clear()
356-
elif line.startswith("Traceback (most recent call last):"):
357-
traceback.append(line)
358-
else:
359-
record(line)
360-
if final and traceback:
361-
record(traceback[-1].strip(), "error", "\n".join(traceback) + "\n")
362-
return b"".join(line + b"\n" for line in out)
363-
364-
365-
def _write_all(fd: int, data: bytes):
366-
"""Write all bytes to a file descriptor.
367-
368-
Args:
369-
fd: The file descriptor.
370-
data: The bytes to write.
371-
"""
372-
view = memoryview(data)
373-
while view:
374-
view = view[os.write(fd, view) :]
375-
376-
377-
class _OutputPump(threading.Thread):
378-
"""Reader thread that turns one child output pipe into JSON lines."""
379-
380-
def __init__(self, read_fd: int, write_fd: int, level: str, name: str):
381-
"""Create the reader.
382-
383-
Args:
384-
read_fd: The read end of the child's output pipe.
385-
write_fd: The file descriptor to write JSON lines to.
386-
level: The record level for plain lines.
387-
name: The logger name for the records.
388-
"""
389-
super().__init__(name=f"reflex-output-{name}", daemon=True)
390-
self.read_fd = read_fd
391-
self.write_fd = write_fd
392-
self.level = level
393-
self.logger_name = name
394-
# When the reader started waiting for data; None while it works.
395-
self.idle_since: float | None = None
396-
# Cumulative write time and the start of any ongoing write.
397-
self._write_state: tuple[float, float | None] = (0.0, None)
398-
self._write_lock = threading.Lock()
399-
400-
def run(self):
401-
"""Forward the pipe until every writer closed it."""
402-
pending = b""
403-
traceback: list[str] = []
404-
while True:
405-
self.idle_since = time.monotonic()
406-
chunk = os.read(self.read_fd, 1 << 16)
407-
self.idle_since = None
408-
if not chunk:
409-
break
410-
data = pending + chunk
411-
# A trailing \r may be the first half of a \r\n.
412-
cut = len(data) - data.endswith(b"\r")
413-
*lines, pending = _LINE_END.split(data[:cut])
414-
pending += data[cut:]
415-
self._write(lines, traceback, final=False)
416-
os.close(self.read_fd)
417-
pending = pending.removesuffix(b"\r")
418-
self._write([pending] if pending else [], traceback, final=True)
419-
420-
def _write(self, lines: list[bytes], traceback: list[str], final: bool):
421-
"""Write converted lines, dropping them if the consumer is gone.
422-
423-
Args:
424-
lines: Complete output lines.
425-
traceback: The lines of a traceback still being collected.
426-
final: Whether the output ended.
427-
"""
428-
data = _to_records(lines, traceback, self.level, self.logger_name, final)
429-
with self._write_lock:
430-
elapsed, _ = self._write_state
431-
started = time.monotonic()
432-
self._write_state = (elapsed, started)
433-
# Keep reading when the consumer is gone: a full pipe blocks the child.
434-
with contextlib.suppress(OSError):
435-
_write_all(self.write_fd, data)
436-
with self._write_lock:
437-
self._write_state = (elapsed + time.monotonic() - started, None)
438-
439-
def writing_snapshot(self) -> tuple[float, float]:
440-
"""Capture the current time and cumulative write duration together.
441-
442-
Returns:
443-
The monotonic time and seconds spent writing up to that time.
444-
"""
445-
with self._write_lock:
446-
elapsed, started = self._write_state
447-
now = time.monotonic()
448-
return now, elapsed if started is None else elapsed + now - started
449-
450-
def drain(self, exited_at: float, writing_at_exit: float, wall_deadline: float):
451-
"""Wait until the pipe is drained after the child exited.
452-
453-
Returns early when the reader sat idle for a while after the exit, or
454-
when the drain took too long. Time spent forwarding output does not
455-
count toward the active limit. A separate wall deadline bounds shutdown
456-
even when a consumer stops reading.
457-
458-
Args:
459-
exited_at: The monotonic time of the pump's snapshot after child exit.
460-
writing_at_exit: The cumulative write time at that snapshot.
461-
wall_deadline: The shared monotonic deadline for shutdown.
462-
"""
463-
while self.is_alive():
464-
self.join(0.05)
465-
now, writing_elapsed = self.writing_snapshot()
466-
idle_since = self.idle_since
467-
# Seconds charged against the five-second active drain budget.
468-
drain_elapsed = now - exited_at - (writing_elapsed - writing_at_exit)
469-
if (
470-
now > wall_deadline
471-
or drain_elapsed > _DRAIN_MAX_SECONDS
472-
or (
473-
idle_since is not None
474-
and now - max(idle_since, exited_at) > _DRAIN_IDLE_SECONDS
475-
)
476-
):
477-
return
478-
479-
480-
def supervise_output(args: list[str]) -> int:
481-
"""Run a command and turn everything it and its descendants print into JSON records.
482-
483-
The command runs with its stdout and stderr on pipes, which every process
484-
it starts inherits. Reader threads write each line to this process's
485-
stdout and stderr as a JSON record: lines that already are JSON log
486-
records pass through, and a Python traceback becomes one record.
487-
488-
Args:
489-
args: The command to run.
490-
491-
Returns:
492-
The command's exit code.
493-
"""
494-
for stream in (sys.stdout, sys.stderr):
495-
if stream is not None:
496-
stream.flush()
497-
# The readers decode UTF-8; Python otherwise uses the locale encoding.
498-
# A configured error handler is kept.
499-
_, _, errors = os.environ.get("PYTHONIOENCODING", "").partition(":")
500-
env = {
501-
"PYTHONUNBUFFERED": "1",
502-
**os.environ,
503-
_SUPERVISED_ENV_VAR: str(os.getpid()),
504-
"PYTHONIOENCODING": f"utf-8:{errors}" if errors else "utf-8",
505-
}
506-
# Raw pipes that only the readers close: closing a pipe while a reader is
507-
# blocked on it waits for that read on Windows.
508-
out_read, out_write = os.pipe()
509-
err_read, err_write = os.pipe()
510-
proc = subprocess.Popen(args, stdout=out_write, stderr=err_write, env=env)
511-
os.close(out_write)
512-
os.close(err_write)
513-
pumps = [
514-
_OutputPump(out_read, 1, "info", "stdout"),
515-
_OutputPump(err_read, 2, "warning", "stderr"),
516-
]
517-
for pump in pumps:
518-
pump.start()
519-
signal.signal(signal.SIGTERM, lambda *_: proc.terminate())
520-
while True:
521-
try:
522-
returncode = proc.wait()
523-
break
524-
except KeyboardInterrupt:
525-
# A terminal Ctrl-C reaches the child too, a SIGINT sent only to
526-
# this process does not. The child handles a repeated interrupt.
527-
if sys.platform != "win32":
528-
proc.send_signal(signal.SIGINT)
529-
wall_deadline = time.monotonic() + _DRAIN_WALL_SECONDS
530-
snapshots = [pump.writing_snapshot() for pump in pumps]
531-
for pump, (exited_at, writing_at_exit) in zip(pumps, snapshots, strict=True):
532-
pump.drain(exited_at, writing_at_exit, wall_deadline)
533-
# A child killed by a signal reports -signum; shells report 128 + signum.
534-
return returncode if returncode >= 0 else 128 - returncode
535-
536-
537263
class JsonHandler(logging.Handler):
538264
"""Emit one JSON object per record for machine consumption."""
539265

0 commit comments

Comments
 (0)