archivebox.core.shutdown_util

Module Contents

Classes

ShutdownSignalState

Tracks the exact OS signal that asked a foreground command to exit.

Functions

raise_if_shutdown_requested	Let long foreground loops honor a signal even if Python ignored it once.
configured_stopwaitsecs	Return a deterministic shutdown bound from generated worker definitions.
wait_popen_and_kill_children	Wait for a Popen parent and then hard-kill any surviving descendants.
wait_psutil_and_kill_children	Wait for a psutil parent and then hard-kill any surviving descendants.
kill_remaining_processes
foreground_shutdown_signals	Install foreground signal handlers that print an immediate exit notice.
foreground_parent_watchdog	Ask a foreground command to exit if its launcher/wrapper disappears.

Data

_active_shutdown_state

API

class archivebox.core.shutdown_util.ShutdownSignalState

Tracks the exact OS signal that asked a foreground command to exit.

signal_name: str | None

None

archivebox.core.shutdown_util._active_shutdown_state: archivebox.core.shutdown_util.ShutdownSignalState | None

None

archivebox.core.shutdown_util.raise_if_shutdown_requested() → None

Let long foreground loops honor a signal even if Python ignored it once.

archivebox.core.shutdown_util.configured_stopwaitsecs(workers: list[dict[str, str]] | tuple[dict[str, str], ...], *, default: int = 5, buffer: int = 5) → int

Return a deterministic shutdown bound from generated worker definitions.

archivebox.core.shutdown_util.wait_popen_and_kill_children(proc: subprocess.Popen, children: list[psutil.Process], *, timeout: float, kill_timeout: float = 2.0) → None

Wait for a Popen parent and then hard-kill any surviving descendants.

archivebox.core.shutdown_util.wait_psutil_and_kill_children(proc: psutil.Process, children: list[psutil.Process], *, timeout: float, kill_timeout: float = 2.0) → None

Wait for a psutil parent and then hard-kill any surviving descendants.

archivebox.core.shutdown_util.kill_remaining_processes(processes: list[psutil.Process], *, timeout: float = 2.0) → None

archivebox.core.shutdown_util.foreground_shutdown_signals(handled_signals: tuple[signal.Signals, ...] = (signal.SIGHUP, signal.SIGINT, signal.SIGTERM), *, first_signal_message: str | None = '\n[🛑] Got {signal_name}, stopping gracefully...\n', on_signal: collections.abc.Callable[[signal.Signals], None] | None = None, raise_on_first_signal: bool = True) → collections.abc.Iterator[archivebox.core.shutdown_util.ShutdownSignalState]

Install foreground signal handlers that print an immediate exit notice.

Some log-tail loops intentionally swallow KeyboardInterrupt so that callers can centralize cleanup in finally blocks. The handler writes the signal name immediately, then raises KeyboardInterrupt to break out of the blocking read.

archivebox.core.shutdown_util.foreground_parent_watchdog(*, enabled: bool = True, check_interval: float = 2.0, shutdown_signal: signal.Signals = signal.SIGTERM) → collections.abc.Iterator[None]

Ask a foreground command to exit if its launcher/wrapper disappears.

uv run archivebox ... and similar wrappers can be killed without delivering a signal to the real Python child. If that child keeps crawling as an orphan, it can hold SQLite write locks long after the user-facing command timed out. This watchdog is only for foreground command lifetimes; daemon/supervisord workers should not use it because their parent may intentionally hand them off.