diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 410e851..2c99873 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -20,7 +20,7 @@ jobs: strategy: fail-fast: false matrix: - python: ['3.10', '3.11', '3.12', '3.13', '3.14'] + python: ['3.10', '3.11', '3.12', '3.13', '3.14', '3.15'] env: UV_PYTHON: ${{ matrix.python }} steps: @@ -46,6 +46,10 @@ jobs: fail-fast: false matrix: include: + - pg: 13 + python: '3.11' + - pg: 14 + python: '3.15' - pg: 15 python: '3.10' - pg: 16 diff --git a/CHANGELOG.md b/CHANGELOG.md index a52d947..4116e24 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,8 +2,36 @@ Notable user-facing changes. -## [Unreleased] - +## [0.5.1] - 2026-10-31 + +- Support Python 3.10–3.15 and PostgreSQL 13–18. Python 3.15 and PostgreSQL 13–14 + needed no runtime change. The clone backend still requires PostgreSQL 15 or later and + now stops with a version error before any dump or restore on older servers. The CI + matrix gains Python 3.15 and PostgreSQL 13 and 14. Version-dependent tests are skipped + on servers that lack the feature, and a flaky test that left a database behind now + cleans up after itself. +- Redraw the project icon: a dashed source database reduced to a small one that keeps the + highlighted row. +- Report `database_stats.already_absent` for databases that were gone when cleanup ran. +- Handle SIGTERM like SIGINT in `dbreduce reduce`: stop new candidates and oracle calls, + stop owned subprocess groups (SIGTERM, then SIGKILL after at most 5 s), remove the + workspace and clone databases and the temporary dump directory, then exit with status + 128+signal. Previously SIGTERM left the workspace database and a full dump behind. +- Bound cancellation by one 30-second shutdown budget shared by subprocess stops, + connections, cleanup SQL and lock waits; a watchdog ends the process shortly after it + and reports what is unconfirmed. A repeated signal during cleanup is ignored; SIGQUIT + or SIGKILL abandons it. +- Report cleanup honestly: removed, already absent, not removed, ownership unverified or + result unknown, per database, separately from the temporary directory, with a way to + check each one. Exit status 1 means cancelled with cleanup unconfirmed. Ownership checks + (OID, owner, run marker) are unchanged, and nothing is removed by name prefix. +- Resolve an interrupted `CREATE DATABASE` by cancelling the statement this run left + on the server before deciding; a database that appears without a marker is reported, + never dropped by name. +- Run `pg_dump`/`pg_restore` and the oracle's matcher in their own process groups so + a stopped run cannot be held by a grandchild that keeps the output pipes open. + The oracle supervisor no longer kills the group on SIGTERM itself; DBReduce sends + SIGTERM, waits a bounded grace period and escalates. - Add repository editing/Git defaults, contribution and security policies, and issue/PR templates; document local checks and dependency-update cadence. - Bound CI job runtimes while retaining the existing test and release matrices. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 4cbf2f4..7cf1717 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -67,7 +67,7 @@ superuser test role. Never point these tests at a shared or production server. DBREDUCE_TEST_ADMIN=postgresql://localhost/postgres uv run --locked pytest -q -m postgres ``` -The existing CI matrix tests Python 3.10–3.14 and PostgreSQL 15–18, builds wheel +The CI matrix tests Python 3.10–3.15 and PostgreSQL 13–18, builds wheel and source distributions, and checks standalone installation. Release publication runs only for version tags or an explicit release workflow dispatch. diff --git a/README.md b/README.md index ff5d9ba..04014a0 100644 --- a/README.md +++ b/README.md @@ -23,7 +23,8 @@ and explicitly configured application relationships. ## Quick start -DBReduce v0.5.0 supports Python 3.10–3.14 and PostgreSQL 15–18. Use current patched, +DBReduce supports Python 3.10–3.15 and PostgreSQL 13–18 (the experimental clone +backend needs PostgreSQL 15 or later). Use current patched, matching-major `pg_dump`, `pg_restore` and `psql` tools; see the [compatibility report](docs/compatibility.md). @@ -65,9 +66,10 @@ target-bug disappearance and infrastructure failures. See the [candidate-validity contract](docs/candidate-invalid.md) and guide for details. Use `--oracle-framed-json` when the application logs to stdout. Snapshot isolation -is the default; `--candidate-backend clone` is an experimental alternative on -PostgreSQL 15–18. Both `WAL_LOG` and `FILE_COPY` are tested; `file_copy_method` -requires PostgreSQL 18. `--restore-jobs N` enables parallel archive restores. +is the default and works on PostgreSQL 13–18; `--candidate-backend clone` is an +experimental alternative on PostgreSQL 15–18. Both `WAL_LOG` and `FILE_COPY` are +tested; `file_copy_method` requires PostgreSQL 18. `--restore-jobs N` enables parallel +archive restores. Use `--reduce-schema` for opt-in whole-table schema reduction. It runs after row reduction, checks each proposed removal with diff --git a/docs/adoption.md b/docs/adoption.md index be60927..157c347 100644 --- a/docs/adoption.md +++ b/docs/adoption.md @@ -45,7 +45,7 @@ v0.5.0 supports the [candidate-validity contract](candidate-invalid.md) for both strict and framed JSON oracles. Case-specific wrappers may explicitly mark known invalid schema/data states; unknown application/environment failures remain infrastructure errors. Current feedback requirements are Python >=3.10, -PostgreSQL 15–18 with matching clients, and DBReduce v0.5.0 or the latest release. +PostgreSQL 13–18 with matching clients, and DBReduce v0.5.1 or the latest release. Reports should state how invalidity was established, separately from negative target checks and unresolved errors. diff --git a/docs/compatibility.md b/docs/compatibility.md index 6bd2675..2003f5f 100644 --- a/docs/compatibility.md +++ b/docs/compatibility.md @@ -5,6 +5,12 @@ the checks performed on 2026-09-30 for DBReduce v0.5.0. This release retains the Python and PostgreSQL compatibility established in v0.4.1 and adds explicit candidate validity without new reducer primitives. +## Range in v0.5.1 + +DBReduce v0.5.1 supports **Python 3.10–3.15** and **PostgreSQL 13–18**. The +[v0.5.1 verification](#v051-verification) below extends the v0.5.0 results; the tables +that follow remain the record of the v0.5.0 validation. + ## Python DBReduce v0.5.0 requires **Python 3.10 or newer**. Each Python version below passed @@ -83,14 +89,52 @@ Servers older than PG15 receive a version error before clone creation. `--file-copy-method` still requires PostgreSQL 18; explicit `copy` requests were rejected clearly on real PG15, PG16 and PG17 servers. -PostgreSQL 14 is **unsupported by policy** for snapshot/schema, not claimed -incompatible. No full snapshot/schema suite was run on 14. Its clone backend is -**known incompatible**: PostgreSQL 14.24 rejects `CREATE DATABASE ... STRATEGY -WAL_LOG` with `option "strategy" not recognized`, and DBReduce rejects it before -creating a clone. PG14 is outside the tested support range and -[reaches end of support on 2026-11-12](https://www.postgresql.org/support/versioning/). -Snapshot mode does not reject PG14 solely on its version, but its -behavior there has not been validated. +Servers older than PostgreSQL 15 cannot use the clone backend: PostgreSQL 14.24 +rejects `CREATE DATABASE ... STRATEGY WAL_LOG` with `option "strategy" not +recognized`. Since v0.5.1 DBReduce stops with a version error before any dump or +restore when `--candidate-backend clone` meets such a server. Snapshot and schema +reduction need no PostgreSQL 15 feature and are verified on 13 and 14 (see +[v0.5.1 verification](#v051-verification)). PostgreSQL +[13 is past end of support and 14 reaches it on 2026-11-12](https://www.postgresql.org/support/versioning/); +use a patched release. + +## v0.5.1 verification + +Checked on 2026-10-31 on macOS arm64 (Python source checks) and Linux arm64 containers +(PostgreSQL), with the committed `uv.lock`. GitHub Actions was not run for these +results; the CI matrix was extended to the same combinations. + +Python 3.10, 3.11, 3.12, 3.13, 3.14 and 3.15 (3.15.2) each passed Ruff, mypy and the +122 non-PostgreSQL tests (also with `DeprecationWarning` as an error), built the wheel +and sdist once, and ran `dbreduce --help` from that wheel through +`uvx --isolated --python `. No source change was needed for 3.15; the locked +psycopg-binary and Typer install and run on it. The wheel declares +`Requires-Python: >=3.10` with no upper bound. + +PostgreSQL integration suite (`pytest -m postgres`) with native, matching-major clients: + +| Server and client | Result | Notes | +| ----------------- | --------------------------- | ------------------------------------------- | +| 13.23 | 45 passed, 21 skipped | snapshot, schema, final restore | +| 14.24 | 45 passed, 21 skipped | snapshot, schema, final restore | +| 15.19 | 65 passed, 1 skipped | adds clone WAL_LOG and FILE_COPY | +| 16.15 | 65 passed, 1 skipped | | +| 17.11 | 65 passed, 1 skipped | | +| 18.6 | 65 passed, 1 skipped | `file_copy_method` is PostgreSQL 18 only | + +The skips are deliberate: clone tests and the per-database ICU provider test need +PostgreSQL 15, and the test that checks the clone version error needs a server older +than 15. Everything else, including cancellation and cleanup, ran on all six majors, +and no run left a `dbreduce_` database behind. The whole suite (167 passed, 21 +skipped) also passed on Python 3.15.2 with PostgreSQL 14.24. Only that pairing and the +Python 3.11 container used for the other majors were executed here; other +Python/PostgreSQL combinations rely on the CI matrix. + +PostgreSQL 13 and 14 have no `CREATE DATABASE ... STRATEGY`, per-database ICU locale +providers or `file_copy_method`. Restoring an archive whose database uses the ICU provider +into such a server fails with the server's own error. `pg_dump --restrict-key`, which +snapshot fingerprinting needs, is present in the 13.23 and 14.24 clients used here; for 13 the +first patched client version was not checked. ## PostgreSQL feature audit @@ -151,9 +195,9 @@ Cross-major combinations were not added to the support matrix. ## CI and installation -The workflow separates five Python unit/type/lint/install jobs from four -PostgreSQL integration jobs (PG15/Python 3.10, PG16/3.12, PG17/3.13, PG18/3.14), -not a 5×4 cross product. Each PostgreSQL job checks all three native client major +The workflow separates six Python unit/type/lint/install jobs from six +PostgreSQL integration jobs (PG13/Python 3.11, PG14/3.15, PG15/3.10, PG16/3.12, +PG17/3.13, PG18/3.14), not a cross product. Each PostgreSQL job checks all three native client major versions, server major, privileged test role and `--restrict-key` availability. The release job depends on both matrices. The tables above summarize v0.5.0 validation; the candidate-validity, oracle-write isolation and final confirmation diff --git a/docs/dbreduce-icon.svg b/docs/dbreduce-icon.svg index b30977c..be0ee0a 100644 --- a/docs/dbreduce-icon.svg +++ b/docs/dbreduce-icon.svg @@ -7,21 +7,26 @@ DBReduce A relational graph reduced while retaining its highlighted node. + >A large dashed database cylinder reduced to a small solid one that keeps a single highlighted row. - - - - - - - - + + + + + + + + + diff --git a/docs/guide.md b/docs/guide.md index c23e4a9..8c5cb5a 100644 --- a/docs/guide.md +++ b/docs/guide.md @@ -13,11 +13,12 @@ python -m pip install dbreduce dbreduce --help ``` -DBReduce v0.5.0 requires Python 3.10 or newer and is tested on Python 3.10–3.14. +DBReduce requires Python 3.10 or newer and is tested on Python 3.10–3.15. See the [v0.5.0 release](https://github.com/0then0/dbreduce/releases/tag/v0.5.0) and [changelog](../CHANGELOG.md) for the candidate-validity protocol extension. -The tested PostgreSQL range is 15–18, separately verified -for snapshot, clone, schema reduction and final logical restore. +The tested PostgreSQL range is 13–18. Snapshot, schema reduction and final logical +restore are verified on all of them; the clone backend needs PostgreSQL 15 or later and +is verified on 15–18. See [the compatibility decision record](compatibility.md). Prefer `uv tool install dbreduce` and check `dbreduce --help`. @@ -219,7 +220,8 @@ and signal conventions, including application exit codes in that range. A wrappe that hides a signal behind another exit code cannot be detected reliably. `--timeout` defaults to 60 seconds for each oracle execution, including failure matcher evaluation. Remaining oracle process-group children are killed after each -execution, including timeout and Ctrl-C. +execution, including timeout and cancellation (see +[Cancellation and cleanup](#cancellation-and-cleanup)). Without a matcher or structured mode, legacy mode warns and accepts any nonzero exit below 126. DB connection failures and other application setup errors cannot be @@ -255,8 +257,8 @@ are rejected because `pg_restore` refuses to read their database settings. Plain SQL is an output format, not an accepted input format. This avoids executing `psql` reconnect/shell meta-commands while restoring user input. -Candidate isolation defaults to `--candidate-backend snapshot`. PostgreSQL 15–18 -can explicitly use `--candidate-backend clone`; see [isolation](isolation.md) for +Candidate isolation defaults to `--candidate-backend snapshot`, available on +PostgreSQL 13–18. PostgreSQL 15–18 can explicitly use `--candidate-backend clone`; see [isolation](isolation.md) for its ownership checks, replication preflight and final logical normalization. `--restore-jobs 4` enables parallel custom-archive restores when the environment benefits; the default remains 1. Clone strategies are `wal_log` (default) and @@ -354,7 +356,7 @@ rejections and cache hits. Accepted/rejected counts partition those completed pr includes all work. `performance.phases` gives count, total and average for each operation. Some phases are nested, such as `restore` within oracle state preparation, so their totals must not be added together. `database_stats` records exact created, -dropped, clone attempts/failures and cleanup failures. `cache_stats` records hit rate +dropped or already absent, clone attempts/failures and cleanup failures. `cache_stats` records hit rate and an estimated time saved from observed oracle cycle cost; it is an estimate, not an observed counterfactual. Clone mode disables the exact fingerprint cache. `database_clone_seconds` is zero for the snapshot backend. Aborted runs publish no @@ -404,6 +406,80 @@ Full restore per probe remains conservative and expensive. See the unsuitable for very large databases: row identities, FK closure and candidate dumps require memory/disk proportional to the dataset. +## Cancellation and cleanup + +`dbreduce reduce` handles **SIGINT** (Ctrl-C) and **SIGTERM** on POSIX systems. The +handlers exist only while the command runs in the main thread and the previous ones +are restored afterwards; importing DBReduce or running `dbreduce inspect` installs none. + +On the first signal DBReduce + +1. records the request and stops: no new candidate and no new oracle call starts; +2. stops the processes it started (the oracle's process group, `pg_dump`, `pg_restore`): + SIGTERM to the group, at most 5 seconds to exit, then SIGKILL to that group and a + reap. Processes DBReduce did not start are never signalled; +3. removes what the run owns: the workspace database, clone databases and the temporary + directory with its dumps. A database is dropped only after its exact name, OID, owner + and run marker match what DBReduce created. Nothing is removed by the `dbreduce_` + prefix, and the source database is never touched; +4. exits without publishing `--output` or `--report`. A cancelled run is never + reported as reproduced, not reproduced or candidate-invalid. + +The signal handler only starts this sequence. Cleanup runs afterwards, not inside it. + +**Shutdown budget.** Steps 2 and 3 share one budget of **30 seconds** from the first +signal. It shortens process waits, connection attempts, cleanup SQL (`statement_timeout`) +and lock waits to the time left, and it is not multiplied by the number of databases. A +watchdog ends the process at most 2 seconds after the budget, even if a call is stuck. The +budget is fixed, not an option. When running in a container, set the stop timeout above it +(for example `docker stop --time 40`; Docker's default is 10 seconds). + +**Repeated signals.** While cleanup runs, further SIGINT or SIGTERM are ignored and noted +on stderr, so one impatient Ctrl-C cannot skip a `finally`. To abandon cleanup at once send +**SIGQUIT** (Ctrl-\) or SIGKILL. Databases and the temporary directory can then remain. + +**Exit status and messages.** Status `130` (SIGINT) or `143` (SIGTERM) means cancelled and +every owned resource was confirmed removed or already absent. Status `1` after a +`Cancelled by ...` message means cancelled but cleanup is not confirmed; stderr lists each +remaining resource with its state: + +- `removed` or `already absent`: nothing remains; +- `not removed`: PostgreSQL refused or timed out; the database is still there; +- `ownership unverified`: a database of that name exists but its OID, owner or marker + differ, so DBReduce left it alone; +- `result unknown`: the server could not be reached, the budget ended, or a reply was + lost; the database may still exist; +- `creation outcome unverified`: the run stopped between `CREATE DATABASE` and writing + its marker. DBReduce first cancels the statement it left running, then reports a + database that exists without a marker instead of dropping it by name. + +Databases and the temporary directory are reported separately: with an unreachable server +the directory with the dump is still removed locally while the databases are listed as +unknown. + +**If cleanup is not confirmed**, check only the databases listed. For each one run the +printed query, `SELECT oid, shobj_description(oid, 'pg_database') FROM pg_database WHERE +datname = ''`. The database belongs to the run if the comment equals the printed +marker (a `creation outcome unverified` database has no comment yet; its name is random +and was generated by the run). Then `DROP DATABASE ""`. Do not clean by pattern. A +listed temporary directory holds a copy of the source data; delete it when no `dbreduce` +process is running. + +**Publication.** Both result files are linked into place only after the working databases +were dropped and the final identity check passed. A signal before that rolls the files +back; once both exist the result is complete, is kept, and a later signal only prints a +note. + +**Not covered.** SIGHUP (a closed terminal or dropped SSH session), SIGKILL, an +out-of-memory kill, a machine failure, or a container +destroyed after its grace period run no cleanup and print nothing; the run's databases +and directory remain, later runs do not look for them, and there is no janitor. Commands +that leave the oracle's process group (daemons, `docker exec` into another container) +are not stopped. The automated checks signal the DBReduce process, or its process +group, directly; they do not reproduce `docker stop` or a CI service's cancellation. +`docker stop` signals only the container's main process, so that process must be DBReduce +or an init that forwards SIGTERM (for example `docker run --init`). + ## Boundaries and limitations - PostgreSQL only; POSIX systems (Linux/macOS) for oracle process-group cleanup. @@ -430,10 +506,11 @@ FK closure and candidate dumps require memory/disk proportional to the dataset. - Do not allow other clients to write into the workspace. Do not use production as a workspace. Clone cleanup verifies exact name, OID, owner and run marker before dropping each owned database. -- Ordinary failures and Ctrl-C attempt to clean up owned databases. An uncertain - clone CREATE reply, killed process or server outage can leave a generated database. - DBReduce reports exact names when it cannot confirm ownership or cleanup; inspect - them manually. It does not force-disconnect other sessions from clone databases. +- Ordinary failures, Ctrl-C and SIGTERM attempt to clean up owned databases within the + [shutdown budget](#cancellation-and-cleanup). An uncertain `CREATE DATABASE` reply, a + killed process (SIGKILL, OOM, container removal) or a server outage can leave a generated + database. DBReduce reports exact names when it can, never after SIGKILL; inspect them + manually. It does not force-disconnect other sessions from clone databases. - Dumps/reports contain application data. Store them appropriately; no DSN is saved in the report. PostgreSQL errors are intentionally summarized without credentials. @@ -458,6 +535,12 @@ duplicate rows, extension-owned tables, locale, and SQL round-trip. The 12,005-r example is intended for a manual end-to-end run; full reduction is deliberately not part of the default test suite. +`tests/test_cancellation_postgres.py` starts real `dbreduce reduce` processes, signals them +at observed stages and checks that owned databases, files and processes are gone while the +source and a foreign database stay intact. Use a dedicated disposable server for it (a +container is enough): its cleanup drops any new `dbreduce_<32 hex>` database that appears +while a test runs, and its process checks read `/proc` (skipped elsewhere). + ## Module interfaces - `reducer.engine.Backend`: `state()` and `attempt(table, rows)`; no PostgreSQL imports. @@ -466,3 +549,5 @@ full reduction is deliberately not part of the default test suite. - `cache`: fingerprints and per-run outcomes. - `graph`: dependencies and strongly connected components for inspection. - `cli`: input validation, orchestration, progress and result files. +- `cancellation`: signal handling, the shutdown budget and the report of unremoved resources. +- `processes`: subprocesses in their own process group, with a bounded stop. diff --git a/docs/isolation.md b/docs/isolation.md index f0b2000..492150c 100644 --- a/docs/isolation.md +++ b/docs/isolation.md @@ -23,8 +23,10 @@ candidate itself is promoted; DBReduce does not replay the deletion. Before dropping a database, DBReduce verifies its generated name, OID, owner, and per-run marker. It does not use a name prefix as ownership proof or use `DROP DATABASE ... WITH (FORCE)` in clone mode. If safe cleanup fails, DBReduce -reports the exact database for inspection. A hard process kill can leave owned -databases behind; inspect ownership before manual cleanup. +reports the exact database for inspection. SIGINT and SIGTERM run the same cleanup +inside a [30-second shutdown budget](guide.md#cancellation-and-cleanup); a hard process +kill (SIGKILL, container removal) can leave owned databases behind. Inspect ownership +before manual cleanup. Clone preflight checks the live source before dumping and the owned initial database after restore. It rejects publications, subscriptions, logical diff --git a/pyproject.toml b/pyproject.toml index f69e88a..507c8be 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "dbreduce" -version = "0.5.0" +version = "0.5.1" description = "Reduce PostgreSQL bug reproducers using delta debugging" readme = "README.md" requires-python = ">=3.10" diff --git a/src/dbreduce/__init__.py b/src/dbreduce/__init__.py index 996845d..f491d92 100644 --- a/src/dbreduce/__init__.py +++ b/src/dbreduce/__init__.py @@ -1,3 +1,3 @@ """PostgreSQL reproducer reduction.""" -__version__ = "0.5.0" +__version__ = "0.5.1" diff --git a/src/dbreduce/cancellation.py b/src/dbreduce/cancellation.py new file mode 100644 index 0000000..3671132 --- /dev/null +++ b/src/dbreduce/cancellation.py @@ -0,0 +1,258 @@ +"""Cooperative cancellation and a finite shutdown budget for one CLI run. + +A signal handler never cleans up. It records the request, starts the shutdown clock and +raises ``Cancelled`` (a ``KeyboardInterrupt``) so the existing ``finally`` and ``__exit__`` +blocks unwind and release what the run owns. Cleanup code runs inside ``protected()`` so a +repeated signal cannot abort it. A watchdog thread bounds the whole shutdown even when a +blocking call never returns. +""" + +import os +import signal +import threading +import time +from collections import Counter +from collections.abc import Callable, Iterator +from contextlib import contextmanager +from dataclasses import dataclass +from types import FrameType +from typing import Any + +# Seconds from the first cancelling signal until the process exits. It covers stopping +# owned subprocesses and every cleanup step. Docker's default stop timeout is 10 seconds, +# so containers need ``--stop-timeout`` above this value to let cleanup finish. +SHUTDOWN_BUDGET = 30.0 +# Extra seconds after the budget for the watchdog to write the report before it forces exit. +REPORT_MARGIN = 2.0 +# Upper bound between SIGTERM and SIGKILL for a process group started by this run. +TERMINATE_GRACE = 5.0 + +CANCEL_SIGNALS = (signal.SIGINT, signal.SIGTERM) + + +class Cancelled(KeyboardInterrupt): + """Raised in the main thread when a cancelling signal arrives.""" + + def __init__(self, signum: int) -> None: + super().__init__(signal.Signals(signum).name) + self.signum = signum + + +@dataclass(frozen=True) +class Leftover: + """A resource owned by this run whose removal is not confirmed.""" + + kind: str # "database" or "directory" + name: str + state: str + detail: str = "" + + +def _notice(message: str) -> None: + # os.write is safe inside a signal handler; buffered stderr may be mid-write. + try: + os.write(2, (message + "\n").encode("utf-8", errors="replace")) + except OSError: + pass + + +def describe_leftovers(leftovers: list[Leftover]) -> list[str]: + lines = [] + for item in leftovers: + lines.append(f" {item.kind} {item.name}: {item.state}") + if item.detail: + lines.append(f" {item.detail}") + return lines + + +class RunControl: + """Signal state, shutdown deadline and the resources a run must report.""" + + def __init__(self, budget: float | None = None) -> None: + self.budget = SHUTDOWN_BUDGET if budget is None else budget + self.requested: int | None = None + self.deadline: float | None = None + self.committed = False + self.ignored_signals = 0 + self.cleaned: Counter[str] = Counter() + self._delivered = False + self._protected = 0 + self._active = False + self._sources: list[Callable[[], list[Leftover]]] = [] + self._groups: set[int] = set() + self._previous: dict[int, Any] = {} + self._signalled = threading.Event() + self._finished = threading.Event() + self._exit_lock = threading.Lock() + self._watchdog: threading.Thread | None = None + + # -- lifecycle ------------------------------------------------------------------- + + def start(self) -> None: + """Install handlers and the watchdog. Does nothing outside the main thread.""" + global _ACTIVE + if _ACTIVE is not None: + return + _ACTIVE = self + self._active = True + if threading.current_thread() is not threading.main_thread(): + return + for number in CANCEL_SIGNALS: + if signal.getsignal(number) == signal.SIG_IGN: + continue # inherited "ignore" (nohup, background job) is the caller's choice + self._previous[number] = signal.signal(number, self._handle) + self._watchdog = threading.Thread(target=self._watch, name="dbreduce-watchdog", daemon=True) + self._watchdog.start() + + def finish(self) -> None: + """Mark cooperative shutdown as done so the watchdog stands down.""" + with self._exit_lock: + self._finished.set() + + def stop(self) -> None: + """Restore the handlers that were installed before ``start``.""" + global _ACTIVE + self.finish() + for number, handler in self._previous.items(): + signal.signal(number, handler) + self._previous.clear() + if _ACTIVE is self: + _ACTIVE = None + self._active = False + + # -- signal handling ------------------------------------------------------------- + + @property + def cancelled(self) -> bool: + return self.requested is not None + + @property + def signal_name(self) -> str: + return signal.Signals(self.requested).name if self.requested is not None else "" + + def _handle(self, signum: int, _frame: FrameType | None) -> None: + if self.requested is not None: + self.ignored_signals += 1 + _notice( + f"Received {signal.Signals(signum).name} again; cleanup is already running and " + "is not interrupted. To abandon it at once send SIGQUIT (Ctrl-\\) or SIGKILL; " + "owned resources may then remain." + ) + return + self.requested = signum + self.deadline = time.monotonic() + self.budget + self._signalled.set() + _notice( + f"Received {signal.Signals(signum).name}: stopping and cleaning up resources owned " + f"by this run (at most {self.budget:g} s)." + ) + if self.committed or self._protected: + return + self._delivered = True + raise Cancelled(signum) + + @contextmanager + def protected(self) -> Iterator[None]: + """Defer cancellation while releasing a resource; deliver it on leaving the section.""" + self._protected += 1 + try: + yield + finally: + self._protected -= 1 + if ( + not self._protected + and self.requested is not None + and not self._delivered + and not self.committed + ): + self._delivered = True + raise Cancelled(self.requested) + + def checkpoint(self) -> None: + """Refuse to start new work once cancellation was requested.""" + if self.requested is not None and not self.committed: + self._delivered = True + raise Cancelled(self.requested) + + def commit(self) -> None: + """Past this point the result is published and later signals no longer cancel.""" + if self._active: + self.committed = True + + # -- shutdown budget ------------------------------------------------------------- + + def remaining(self) -> float | None: + return None if self.deadline is None else self.deadline - time.monotonic() + + def expired(self) -> bool: + left = self.remaining() + return left is not None and left <= 0 + + def bounded(self, default: float) -> float: + """Shrink a timeout to the time left in the shutdown budget (at least one second).""" + left = self.remaining() + return default if left is None else max(1.0, min(default, left)) + + # -- ownership reporting --------------------------------------------------------- + + def register(self, source: Callable[[], list[Leftover]]) -> None: + if self._active: + self._sources.append(source) + + def leftovers(self) -> list[Leftover]: + found: list[Leftover] = [] + for source in self._sources: + try: + found.extend(source()) + except Exception: + found.append(Leftover("resource", "unknown", "could not be inspected")) + return found + + def record_cleanup(self, status: str) -> None: + """Count confirmed removals that happened after cancellation was requested.""" + if self.requested is not None: + self.cleaned[status] += 1 + + def track_group(self, pgid: int) -> None: + self._groups.add(pgid) + + def untrack_group(self, pgid: int) -> None: + self._groups.discard(pgid) + + # -- watchdog -------------------------------------------------------------------- + + def _watch(self) -> None: + # Cancelling signals must reach the main thread, never this one. + if hasattr(signal, "pthread_sigmask"): + signal.pthread_sigmask(signal.SIG_BLOCK, CANCEL_SIGNALS) + self._signalled.wait() + assert self.deadline is not None + if self._finished.wait(max(0.0, self.deadline - time.monotonic()) + REPORT_MARGIN): + return + with self._exit_lock: + if self._finished.is_set(): + return + lines = [ + f"Shutdown budget of {self.budget:g} s exhausted after {self.signal_name}; " + "forcing exit. Cleanup is NOT confirmed." + ] + leftovers = self.leftovers() + if leftovers: + lines.append("These resources owned by this run may remain:") + lines.extend(describe_leftovers(leftovers)) + _notice("\n".join(lines)) + for pgid in list(self._groups): + try: + os.killpg(pgid, signal.SIGKILL) + except OSError: + pass + os._exit(1) + + +_ACTIVE: RunControl | None = None +_IDLE = RunControl() + + +def current() -> RunControl: + """The control of the running CLI command; an inert one for library use.""" + return _ACTIVE or _IDLE diff --git a/src/dbreduce/cli/__init__.py b/src/dbreduce/cli/__init__.py index 1d07292..b07c82a 100644 --- a/src/dbreduce/cli/__init__.py +++ b/src/dbreduce/cli/__init__.py @@ -2,8 +2,10 @@ import os import re import shutil +import signal import tempfile import time +from collections.abc import Callable from contextlib import ExitStack from pathlib import Path from typing import Annotated @@ -13,6 +15,7 @@ from dbreduce import __version__ from dbreduce.cache.store import Cache +from dbreduce.cancellation import Leftover, RunControl, current, describe_leftovers from dbreduce.graph.dependencies import components, dependencies from dbreduce.oracle.runner import Oracle from dbreduce.performance import Performance @@ -36,7 +39,11 @@ def publish_result(exported: Path, output: Path, report: Path, result: dict[str, object]) -> None: - """Prepare both files, then publish without overwriting existing paths.""" + """Prepare both files, then publish without overwriting existing paths. + + Cancellation before the second link rolls the first back. After both links exist the + result is published and a later signal no longer cancels the run. + """ pending: list[Path] = [] created: list[Path] = [] try: @@ -59,13 +66,82 @@ def publish_result(exported: Path, output: Path, report: Path, result: dict[str, for prepared_path, destination in zip(pending, (output, report), strict=True): os.link(prepared_path, destination) created.append(destination) + current().commit() except BaseException: - for path in created: - path.unlink(missing_ok=True) + with current().protected(): + for path in created: + path.unlink(missing_ok=True) raise finally: - for path in pending: - path.unlink(missing_ok=True) + with current().protected(): + for path in pending: + path.unlink(missing_ok=True) + + +def error_summaries(error: BaseException) -> list[str]: + """Distinct messages along an exception chain, without database error details.""" + summaries: list[str] = [] + seen: set[int] = set() + current_error: BaseException | None = error + while current_error is not None and id(current_error) not in seen: + seen.add(id(current_error)) + if not isinstance(current_error, KeyboardInterrupt): + message = ( + "PostgreSQL operation failed; check connection and permissions" + if isinstance(current_error, psycopg.Error) + else str(current_error) or type(current_error).__name__ + ) + if message not in summaries: + summaries.append(message) + current_error = current_error.__cause__ or current_error.__context__ + return summaries + + +def directory_leftovers(path: Path) -> Callable[[], list[Leftover]]: + def leftovers() -> list[Leftover]: + if not path.exists(): + return [] + return [ + Leftover( + "directory", + str(path), + "not removed", + "It may hold a full copy of the source data. Delete it once no dbreduce " + "process is running.", + ) + ] + + return leftovers + + +def report_cancellation( + control: RunControl, error: BaseException, started: float, temporary: Path | None +) -> int: + """Print what happened after a cancelling signal and return the exit status.""" + + def say(message: str) -> None: + typer.echo(message, err=True) + + say( + f"Cancelled by {control.signal_name} after {time.monotonic() - started:.1f} s. " + "No reduction result was published." + ) + for message in error_summaries(error): + say(f" detail: {message}") + leftovers = control.leftovers() + confirmed = [f"{count} database(s) {state}" for state, count in control.cleaned.items()] + if temporary is not None and not temporary.exists(): + confirmed.append("temporary directory removed") + if confirmed: + say("Confirmed after cancellation: " + ", ".join(confirmed) + ".") + if not leftovers: + say("Cleanup complete: no resource owned by this run remains.") + return 128 + int(control.requested or signal.SIGTERM) + say("Cleanup is NOT complete. These resources owned by this run may remain:") + for line in describe_leftovers(leftovers): + say(line) + say("DBReduce does not remove databases by name prefix; check each one listed above.") + return 1 def require_clients() -> None: @@ -159,7 +235,10 @@ def reduce_command( """Reduce a source database or dump; export SQL and a JSON report.""" started = time.monotonic() performance = Performance() + control = RunControl() + temporary_path: Path | None = None try: + control.start() if (database is None) == (input_dump is None) or database == "": raise ValueError("Provide exactly one of --database or --dump") admin = admin_database or database @@ -197,7 +276,17 @@ def reduce_command( err=True, ) require_clients() + if candidate_backend == "clone": + # Fail before any dump or restore on servers without CREATE DATABASE ... STRATEGY. + with psycopg.connect(admin, connect_timeout=10) as conn: + if conn.info.server_version < 150000: + raise ValueError( + "Clone backend requires PostgreSQL 15 or later; " + "use --candidate-backend snapshot" + ) with tempfile.TemporaryDirectory(prefix="dbreduce-") as temporary: + temporary_path = Path(temporary) + control.register(directory_leftovers(temporary_path)) snapshot = Path(temporary) / "accepted.dump" if database: with psycopg.connect(database, connect_timeout=10) as conn: @@ -461,6 +550,8 @@ def reduce_command( "fallbacks": 0, "created": workspace.databases_created + (store.created if store else 0), "dropped": workspace.databases_dropped + (store.dropped if store else 0), + "already_absent": workspace.databases_absent + + (store.already_absent if store else 0), "cleanup_failures": workspace.cleanup_failures + (store.cleanup_failures if store else 0), } @@ -469,6 +560,9 @@ def reduce_command( database_stats["dropped"] = workspace.databases_dropped + ( store.dropped if store is not None else 0 ) + database_stats["already_absent"] = workspace.databases_absent + ( + store.already_absent if store is not None else 0 + ) database_stats["cleanup_failures"] = workspace.cleanup_failures + ( store.cleanup_failures if store is not None else 0 ) @@ -482,5 +576,18 @@ def reduce_command( f"Final: {final_rows} rows, oracle: FAIL, " f"executions: {runner.executions}, cache hits: {cache.hits}" ) - except (OSError, ValueError, re.error, RuntimeError, psycopg.Error) as error: - fail(error) + if control.cancelled: + typer.echo( + f"Note: {control.signal_name} arrived after the result was published; " + "the published result was kept.", + err=True, + ) + except BaseException as error: + control.finish() + if control.cancelled and not control.committed: + raise typer.Exit(report_cancellation(control, error, started, temporary_path)) from None + if isinstance(error, (OSError, ValueError, re.error, RuntimeError, psycopg.Error)): + fail(error) + raise + finally: + control.stop() diff --git a/src/dbreduce/oracle/runner.py b/src/dbreduce/oracle/runner.py index be0b525..14578ff 100644 --- a/src/dbreduce/oracle/runner.py +++ b/src/dbreduce/oracle/runner.py @@ -3,7 +3,6 @@ import os import re import selectors -import signal import subprocess import sys import time @@ -11,7 +10,9 @@ from collections.abc import Callable from enum import Enum +from dbreduce.cancellation import TERMINATE_GRACE, current from dbreduce.performance import Performance +from dbreduce.processes import close_pipes, run_owned, stop_group VERDICT_PREFIX = "DBREDUCE_VERDICT " @@ -45,6 +46,10 @@ class OracleError(RuntimeError): print(json.dumps(matched)) """ +# DBReduce signals the whole process group: SIGTERM first, SIGKILL after a grace period. +# The supervisor only has to outlive the command so it can report its status and then kill +# whatever remains in the group. A no-op handler (not SIG_IGN) keeps the command's own +# SIGTERM disposition at the default, because ignored signals survive exec. _ORACLE_SUPERVISOR = """ import os import signal @@ -53,10 +58,7 @@ class OracleError(RuntimeError): status_fd = int(os.environ.pop("DBREDUCE_ORACLE_STATUS_FD")) -def terminate_group(_signum, _frame): - os.killpg(os.getpgrp(), signal.SIGKILL) - -signal.signal(signal.SIGTERM, terminate_group) +signal.signal(signal.SIGTERM, lambda _signum, _frame: None) child = subprocess.Popen(sys.argv[2], shell=True) status = child.wait() os.write(status_fd, str(status).encode("ascii")) @@ -161,7 +163,7 @@ def _classify( if matcher_timeout <= 0: raise subprocess.TimeoutExpired("oracle matcher", self.timeout) try: - result = subprocess.run( + result = run_owned( [sys.executable, "-c", _REGEX_MATCHER], input=json.dumps( { @@ -172,7 +174,6 @@ def _classify( "texts": [stdout, stderr], } ), - capture_output=True, text=True, timeout=matcher_timeout, check=True, @@ -189,6 +190,7 @@ def _classify( return OracleOutcome.SAME_FAILURE, identity def _run(self, env: dict[str, str]) -> tuple[int, str, str]: + current().checkpoint() capture_out = self.mode in ("json", "framed_json") or self.stdout_pattern is not None capture_err = self.stderr_pattern is not None buffers = [bytearray(), bytearray()] @@ -196,62 +198,73 @@ def _run(self, env: dict[str, str]) -> tuple[int, str, str]: os.set_inheritable(status_write, True) supervisor_env = {**env, "DBREDUCE_ORACLE_STATUS_FD": str(status_write)} deadline = time.monotonic() + self.timeout + process: subprocess.Popen[bytes] | None = None + control = current() try: - with subprocess.Popen( - [sys.executable, "-c", _ORACLE_SUPERVISOR, str(status_write), self.command], - shell=False, - env=supervisor_env, - start_new_session=True, - pass_fds=(status_write,), - stdout=subprocess.PIPE if capture_out else subprocess.DEVNULL, - stderr=subprocess.PIPE if capture_err else subprocess.DEVNULL, - ) as process: - os.close(status_write) - status_data = bytearray() - status_complete = False - try: - with selectors.DefaultSelector() as selector: - selector.register(status_read, selectors.EVENT_READ, "status") - for index, stream in enumerate((process.stdout, process.stderr)): - if stream is not None: - selector.register(stream, selectors.EVENT_READ, index) - while selector.get_map(): - remaining = deadline - time.monotonic() - if remaining <= 0: - raise subprocess.TimeoutExpired(self.command, self.timeout) - for key, _ in selector.select(min(remaining, 0.05)): - chunk = os.read(key.fd, 65536) - if not chunk: - selector.unregister(key.fileobj) - if key.data == "status": - status_complete = True - continue + # A cancelling signal must not land between starting the supervisor and + # knowing about it: the group is registered before the signal can be delivered. + with control.protected(): + process = subprocess.Popen( + [sys.executable, "-c", _ORACLE_SUPERVISOR, str(status_write), self.command], + shell=False, + env=supervisor_env, + start_new_session=True, + pass_fds=(status_write,), + stdout=subprocess.PIPE if capture_out else subprocess.DEVNULL, + stderr=subprocess.PIPE if capture_err else subprocess.DEVNULL, + ) + control.track_group(process.pid) + os.close(status_write) + status_data = bytearray() + status_complete = False + try: + with selectors.DefaultSelector() as selector: + selector.register(status_read, selectors.EVENT_READ, "status") + for index, stream in enumerate((process.stdout, process.stderr)): + if stream is not None: + selector.register(stream, selectors.EVENT_READ, index) + while selector.get_map(): + remaining = deadline - time.monotonic() + if remaining <= 0: + raise subprocess.TimeoutExpired(self.command, self.timeout) + for key, _ in selector.select(min(remaining, 0.05)): + chunk = os.read(key.fd, 65536) + if not chunk: + selector.unregister(key.fileobj) if key.data == "status": - status_data.extend(chunk) - if len(status_data) > 4: - raise OracleError("Invalid oracle supervisor status") - continue - buffer = buffers[key.data] - if len(buffer) + len(chunk) > 1024 * 1024: - raise OracleError("Oracle output exceeds 1 MiB identity limit") - buffer.extend(chunk) - if not status_complete or not status_data: - raise OracleError("Oracle supervisor exited without a command status") - try: - code = int(status_data) - except ValueError as error: - raise OracleError("Invalid oracle supervisor status") from error - process.wait(timeout=max(0, deadline - time.monotonic())) - except BaseException: - # Signal the unreaped supervisor PID. Its handler kills its - # process group while the group leader is still alive. - try: - process.send_signal(signal.SIGTERM) - except ProcessLookupError: - pass - process.wait() - raise + status_complete = True + continue + if key.data == "status": + status_data.extend(chunk) + if len(status_data) > 4: + raise OracleError("Invalid oracle supervisor status") + continue + buffer = buffers[key.data] + if len(buffer) + len(chunk) > 1024 * 1024: + raise OracleError("Oracle output exceeds 1 MiB identity limit") + buffer.extend(chunk) + if not status_complete or not status_data: + raise OracleError("Oracle supervisor exited without a command status") + try: + code = int(status_data) + except ValueError as error: + raise OracleError("Invalid oracle supervisor status") from error + process.wait(timeout=max(0, deadline - time.monotonic())) + except BaseException as error: + # Stop the supervisor's process group: SIGTERM and a bounded grace for + # cancellation, an immediate kill for a timeout or an oracle error. + immediate = isinstance(error, (subprocess.TimeoutExpired, OracleError)) + stop_group(process, grace=0 if immediate else TERMINATE_GRACE) + raise + except BaseException: + # Cancellation delivered right after the supervisor started, before the inner try. + if process is not None and process.poll() is None: + stop_group(process, grace=TERMINATE_GRACE) + raise finally: + if process is not None: + control.untrack_group(process.pid) + close_pipes(process) os.close(status_read) try: os.close(status_write) @@ -270,6 +283,7 @@ def fails( cleanup: Callable[[], None] | None = None, ) -> bool: for _ in range(self.confirm): + current().checkpoint() try: with self.performance.measure("oracle_state_preparation"): prepared_url = prepare() diff --git a/src/dbreduce/postgres/backend.py b/src/dbreduce/postgres/backend.py index dc80431..0b3f834 100644 --- a/src/dbreduce/postgres/backend.py +++ b/src/dbreduce/postgres/backend.py @@ -6,6 +6,7 @@ import psycopg from dbreduce.cache.store import Cache, fingerprint +from dbreduce.cancellation import current from dbreduce.models.schema import RowKey, Schema, State, TableKey from dbreduce.oracle.runner import Oracle from dbreduce.performance import Performance @@ -47,6 +48,7 @@ def state(self) -> State: return self.current def attempt(self, table: TableKey, rows: list[RowKey]) -> bool: + current().checkpoint() self.probes += 1 with self.performance.measure("accepted_restore"): self.workspace.reset(self.snapshot) @@ -151,6 +153,7 @@ def cleanup() -> None: return self.oracle.fails("", prepare, cleanup) def attempt(self, table: TableKey, rows: list[RowKey]) -> bool: + current().checkpoint() self.probes += 1 candidate_db = self.store.clone(self.accepted_db) promoted = False diff --git a/src/dbreduce/postgres/clone.py b/src/dbreduce/postgres/clone.py index c9d30a1..8b33d10 100644 --- a/src/dbreduce/postgres/clone.py +++ b/src/dbreduce/postgres/clone.py @@ -10,12 +10,18 @@ from psycopg import sql from psycopg.conninfo import make_conninfo +from dbreduce.cancellation import Leftover, current from dbreduce.models.schema import Schema, State from dbreduce.performance import Performance from dbreduce.postgres.database import ( + CleanupResult, + CleanupStatus, DatabaseSettings, database_url, + drop_owned_database, initial_database_statement, + inspection_hint, + settle_unproven_create, ) from dbreduce.postgres.dump import restore from dbreduce.postgres.rows import read_rows @@ -79,8 +85,10 @@ def __init__( self.marker = "dbreduce-run:" + uuid.uuid4().hex self.owned: dict[str, OwnedDatabase] = {} self.unproven: set[str] = set() + self.last_cleanup: dict[str, CleanupResult] = {} self.created = 0 self.dropped = 0 + self.already_absent = 0 self.clone_attempts = 0 self.clone_failures = 0 self.cleanup_failures = 0 @@ -97,6 +105,32 @@ def __init__( if file_copy_method is not None and self.server_file_copy_method is None: raise ValueError("file_copy_method requires PostgreSQL 18") self.clone_probe: str | None = None + current().register(self.leftovers) + + def leftovers(self) -> list[Leftover]: + found = [] + for database in list(self.owned.values()): # the watchdog may call this concurrently + result = self.last_cleanup.get(database.name) + state = result.status.value if result is not None else "removal not attempted" + hint = inspection_hint(self.admin_dsn, database.name, self.marker) + found.append( + Leftover( + "database", + database.name, + state, + ". ".join(filter(None, (result.detail if result else "", hint))), + ) + ) + for name in sorted(self.unproven): + found.append( + Leftover( + "database", + name, + "creation outcome unverified", + inspection_hint(self.admin_dsn, name, None), + ) + ) + return found def probe_file_copy_clone(self) -> None: if self.strategy != "file_copy" or ( @@ -123,8 +157,13 @@ def probe_file_copy_clone(self) -> None: if probe is not None: self.drop(probe) - def _admin(self) -> psycopg.Connection[tuple[object, ...]]: - return psycopg.connect(self.admin_dsn, autocommit=True, connect_timeout=10) + def _admin(self, application_name: str | None = None) -> psycopg.Connection[tuple[object, ...]]: + return psycopg.connect( + self.admin_dsn, + autocommit=True, + connect_timeout=10, + application_name=application_name, + ) def dsn(self, database: OwnedDatabase) -> str: return make_conninfo(self.admin_dsn, dbname=database.name) @@ -140,7 +179,8 @@ def _create( if clone: self.clone_attempts += 1 try: - with self._admin() as conn: + # The session is named after the database so an unfinished CREATE can be found. + with self._admin(name) as conn: conn.execute("SET statement_timeout = '600s'") if self.requested_file_copy_method is not None: conn.execute( @@ -152,22 +192,24 @@ def _create( self.unproven.add(name) with self.performance.measure("database_clone" if clone else "database_create"): conn.execute(query) - row = conn.execute( - "SELECT d.oid, d.datdba FROM pg_database d WHERE d.datname = %s", - (name,), - ).fetchone() - if row is None: - raise RuntimeError(f"Cannot verify created database {name}") - conn.execute( - sql.SQL("COMMENT ON DATABASE {} IS {}").format( - sql.Identifier(name), sql.Literal(self.marker) + # A signal must not split recording the database from marking it as ours. + with current().protected(): + row = conn.execute( + "SELECT d.oid, d.datdba FROM pg_database d WHERE d.datname = %s", + (name,), + ).fetchone() + if row is None: + raise RuntimeError(f"Cannot verify created database {name}") + conn.execute( + sql.SQL("COMMENT ON DATABASE {} IS {}").format( + sql.Identifier(name), sql.Literal(self.marker) + ) ) - ) - database = OwnedDatabase(name, cast(int, row[0]), cast(int, row[1])) - self.owned[name] = database - self.unproven.remove(name) - self.created += 1 - return database + database = OwnedDatabase(name, cast(int, row[0]), cast(int, row[1])) + self.owned[name] = database + self.unproven.remove(name) + self.created += 1 + return database except psycopg.errors.DuplicateDatabase: self.unproven.discard(name) if clone: @@ -287,34 +329,55 @@ def clone(self, source: OwnedDatabase) -> OwnedDatabase: ) def drop(self, database: OwnedDatabase) -> None: - try: - self._verify(database) - with self._admin() as conn: - conn.execute("SET statement_timeout = '600s'") - with self.performance.measure("database_cleanup"): - conn.execute(sql.SQL("DROP DATABASE {}").format(sql.Identifier(database.name))) - except RuntimeError: - self.cleanup_failures += 1 - raise - except psycopg.Error as error: + if self.owned.get(database.name) is not database: + raise RuntimeError(f"Database {database.name} is not tracked by this run") + # Cancellation is delivered after the drop, never in the middle of it. + with current().protected(): + result = drop_owned_database( + self.admin_dsn, + database.name, + oid=database.oid, + owner=database.owner, + marker=self.marker, + force=False, + lock_key=None, + performance=self.performance, + ) + self.last_cleanup[database.name] = result + if not result.done: self.cleanup_failures += 1 + if result.status is CleanupStatus.UNVERIFIED: + raise RuntimeError( + f"Ownership of database {database.name} cannot be verified; " + "inspect it manually before removal" + ) raise RuntimeError( - f"Could not clean up database {database.name}; inspect sessions, " - "prepared transactions, logical slots and subscriptions, then remove it manually" - ) from error + f"Could not clean up database {database.name} ({result.detail}); inspect " + "sessions, prepared transactions, logical slots and subscriptions, then " + "remove it manually" + ) from result.error self.owned.pop(database.name) - self.dropped += 1 + current().record_cleanup(result.status.value) + if result.status is CleanupStatus.REMOVED: + self.dropped += 1 + else: + self.already_absent += 1 def close(self) -> None: failures = [] - for database in list(self.owned.values())[::-1]: - try: - self.drop(database) - except (RuntimeError, KeyboardInterrupt) as error: - failures.append( - f"Could not confirm cleanup of database {database.name}: " - f"{str(error) or type(error).__name__}; inspect it before manual removal" - ) + with current().protected(): + for database in list(self.owned.values())[::-1]: + try: + self.drop(database) + except (RuntimeError, KeyboardInterrupt) as error: + failures.append( + f"Could not confirm cleanup of database {database.name}: " + f"{str(error) or type(error).__name__}; inspect it before manual removal" + ) + for name in sorted(self.unproven): + exists = settle_unproven_create(self.admin_dsn, name) + if exists is False: + self.unproven.discard(name) if self.unproven: failures.append("Unverified creation outcome for: " + ", ".join(sorted(self.unproven))) if failures: diff --git a/src/dbreduce/postgres/database.py b/src/dbreduce/postgres/database.py index 31b7d4b..0d31e16 100644 --- a/src/dbreduce/postgres/database.py +++ b/src/dbreduce/postgres/database.py @@ -1,9 +1,12 @@ import hashlib +import math import os import re import subprocess +import time import uuid from dataclasses import dataclass +from enum import Enum from pathlib import Path from types import TracebackType from urllib.parse import quote, urlencode @@ -12,8 +15,10 @@ from psycopg import sql from psycopg.conninfo import conninfo_to_dict, make_conninfo +from dbreduce.cancellation import Leftover, current from dbreduce.performance import Performance from dbreduce.postgres.dump import restore +from dbreduce.processes import run_owned @dataclass(frozen=True) @@ -45,7 +50,7 @@ def read_archive_settings(path: Path) -> DatabaseSettings: """Use pg_restore's own CREATE DATABASE statement as archive metadata.""" with path.open("rb") as archive: try: - result = subprocess.run( + result = run_owned( [ "pg_restore", "--create", @@ -56,10 +61,8 @@ def read_archive_settings(path: Path) -> DatabaseSettings: "-", ], stdin=archive, - capture_output=True, timeout=600, env={**os.environ, "LC_ALL": "C"}, - check=False, ) except subprocess.TimeoutExpired as error: raise ValueError("Timed out reading archive database settings") from error @@ -163,6 +166,182 @@ def initial_database_statement(name: str, settings: DatabaseSettings | None) -> return statement +class CleanupStatus(str, Enum): + """Outcome of removing one database owned by this run.""" + + REMOVED = "removed" # DROP completed; the catalog no longer lists it + ABSENT = "already absent" # nothing to remove + NOT_REMOVED = "not removed" # PostgreSQL refused or timed out; still present + UNVERIFIED = "ownership unverified" # a database with this name differs from the one created + UNKNOWN = "result unknown" # the server could not be asked, or the answer was lost + + +@dataclass(frozen=True) +class CleanupResult: + status: CleanupStatus + detail: str = "" + error: BaseException | None = None + + @property + def done(self) -> bool: + return self.status in (CleanupStatus.REMOVED, CleanupStatus.ABSENT) + + +def statement_timeout(default_seconds: int) -> str: + """SET statement for a cleanup statement, shortened to the remaining shutdown budget.""" + return f"SET statement_timeout = '{math.ceil(current().bounded(default_seconds))}s'" + + +def server_label(dsn: str) -> str: + """host:port of a DSN for reports; never the user, password or database.""" + try: + params = conninfo_to_dict(dsn) + except psycopg.ProgrammingError: + return "" + host = str(params.get("host") or params.get("hostaddr") or "") + port = params.get("port") + return f"{host}:{port}" if host and port else host + + +def inspection_hint(dsn: str, name: str, marker: str | None) -> str: + """How an operator can check one specific database before removing it by hand.""" + where = server_label(dsn) + query = ( + "SELECT oid, shobj_description(oid, 'pg_database') FROM pg_database " + f"WHERE datname = '{name}'" + ) + if marker is None: + proof = ( + "It has no run marker. The name is random and was generated by this run, but " + "DBReduce does not treat a name as proof of ownership." + ) + else: + proof = f"It belongs to this run only if its comment is '{marker}'." + return (f"On {where}: " if where else "") + f"{query}. {proof} Drop it only after checking." + + +def drop_owned_database( + admin_dsn: str, + name: str, + *, + oid: int, + owner: int, + marker: str, + force: bool, + lock_key: int | None, + performance: Performance, +) -> CleanupResult: + """Drop one database after matching its OID, owner and marker; say what is known. + + This is the single removal contract for workspace and clone databases. The result + separates a confirmed drop, an already absent database, a refused drop, a name that no + longer matches what was created, and an outcome that could not be established. + """ + control = current() + if control.expired(): + return CleanupResult( + CleanupStatus.UNKNOWN, "the shutdown budget ended before the drop was attempted" + ) + statement = sql.SQL("DROP DATABASE {}" + (" WITH (FORCE)" if force else "")).format( + sql.Identifier(name) + ) + dropping = False + try: + with psycopg.connect( + admin_dsn, autocommit=True, connect_timeout=int(control.bounded(10)) + ) as conn: + conn.execute(statement_timeout(600)) + if lock_key is not None: + conn.execute("SELECT pg_advisory_lock(%s)", (lock_key,)) + row = conn.execute( + "SELECT oid, datdba, shobj_description(oid, 'pg_database') " + "FROM pg_database WHERE datname = %s", + (name,), + ).fetchone() + if row is None: + return CleanupResult(CleanupStatus.ABSENT) + if (row[0], row[1], row[2]) != (oid, owner, marker): + return CleanupResult( + CleanupStatus.UNVERIFIED, + "a database with this name exists but its OID, owner or marker differ", + ) + dropping = True + with performance.measure("database_cleanup"): + conn.execute(statement) + return CleanupResult(CleanupStatus.REMOVED) + except psycopg.Error as error: + state = getattr(error, "sqlstate", None) + if not dropping: + if state is None: + return CleanupResult( + CleanupStatus.UNKNOWN, "the server could not be reached", error + ) + return CleanupResult( + CleanupStatus.NOT_REMOVED, f"cleanup statement failed (SQLSTATE {state})", error + ) + # The server may have finished the DROP even though the reply was lost. + try: + with psycopg.connect( + admin_dsn, autocommit=True, connect_timeout=int(control.bounded(10)) + ) as conn: + present = conn.execute( + "SELECT oid FROM pg_database WHERE datname = %s", (name,) + ).fetchone() + except psycopg.Error: + return CleanupResult( + CleanupStatus.UNKNOWN, + "the DROP reply was lost and the server could not be checked", + error, + ) + if present is None or present[0] != oid: + return CleanupResult(CleanupStatus.REMOVED, "confirmed by a catalog check", error) + if state is not None: + return CleanupResult( + CleanupStatus.NOT_REMOVED, f"PostgreSQL rejected the DROP (SQLSTATE {state})", error + ) + return CleanupResult( + CleanupStatus.UNKNOWN, + "the DROP reply was lost and the database was still listed when checked", + error, + ) + + +def settle_unproven_create(admin_dsn: str, name: str) -> bool | None: + """Resolve an ambiguous CREATE DATABASE: True if it exists, False if absent, else None. + + The creating session names itself after the database. A statement this run left running + on the server is cancelled first, so the database cannot appear after the answer. + """ + control = current() + try: + with psycopg.connect( + admin_dsn, autocommit=True, connect_timeout=int(control.bounded(10)) + ) as conn: + conn.execute(statement_timeout(30)) + give_up = time.monotonic() + control.bounded(10) + while True: + pids = list( + conn.execute( + "SELECT pid FROM pg_stat_activity " + "WHERE application_name = %s AND pid <> pg_backend_pid()", + (name,), + ).fetchall() + ) + if not pids: + break + for (pid,) in pids: + conn.execute("SELECT pg_cancel_backend(%s)", (pid,)) + if time.monotonic() >= give_up: + return None + time.sleep(0.1) + row = conn.execute( + "SELECT EXISTS (SELECT 1 FROM pg_database WHERE datname = %s)", (name,) + ).fetchone() + except psycopg.Error: + return None + return None if row is None else bool(row[0]) + + class Workspace: """Own exactly one random database; no destructive method accepts a user database name.""" @@ -188,38 +367,68 @@ def __init__( self.unproven = False self.databases_created = 0 self.databases_dropped = 0 + self.databases_absent = 0 self.cleanup_failures = 0 + self.last_cleanup: CleanupResult | None = None self.performance = performance or Performance() self.restore_jobs = restore_jobs + current().register(self.leftovers) def __enter__(self) -> "Workspace": return self + def leftovers(self) -> list[Leftover]: + if self.created: + result = self.last_cleanup + state = result.status.value if result is not None else "removal not attempted" + detail = result.detail if result is not None else "" + hint = inspection_hint(self.admin_dsn, self.name, self.marker) + return [Leftover("database", self.name, state, ". ".join(filter(None, (detail, hint))))] + if self.unproven: + return [ + Leftover( + "database", + self.name, + "creation outcome unverified", + inspection_hint(self.admin_dsn, self.name, None), + ) + ] + return [] + def create(self) -> None: self.close() statement = initial_database_statement(self.name, self.settings) try: - with psycopg.connect(self.admin_dsn, autocommit=True, connect_timeout=10) as conn: + # The session is named after the database so an unfinished CREATE can be found. + with psycopg.connect( + self.admin_dsn, + autocommit=True, + connect_timeout=10, + application_name=self.name, + ) as conn: conn.execute("SET statement_timeout = '600s'") conn.execute("SELECT pg_advisory_lock(%s)", (self.lock_key,)) # A lost CREATE reply is ambiguous. A name alone is never ownership proof. self.unproven = True with self.performance.measure("database_create"): conn.execute(statement) - row = conn.execute( - "SELECT oid, datdba FROM pg_database WHERE datname = %s", (self.name,) - ).fetchone() - if row is None: - raise RuntimeError(f"Cannot verify created workspace database {self.name}") - conn.execute( - sql.SQL("COMMENT ON DATABASE {} IS {}").format( - sql.Identifier(self.name), sql.Literal(self.marker) + # A signal must not split recording the database from marking it as ours. + with current().protected(): + row = conn.execute( + "SELECT oid, datdba FROM pg_database WHERE datname = %s", (self.name,) + ).fetchone() + if row is None: + raise RuntimeError(f"Cannot verify created workspace database {self.name}") + conn.execute( + sql.SQL("COMMENT ON DATABASE {} IS {}").format( + sql.Identifier(self.name), sql.Literal(self.marker) + ) ) - ) - self.oid, self.owner = int(row[0]), int(row[1]) - self.created = True - self.unproven = False - self.databases_created += 1 + self.oid, self.owner = int(row[0]), int(row[1]) + self.created = True + self.unproven = False + self.last_cleanup = None # a result about an earlier database + self.databases_created += 1 except psycopg.errors.DuplicateDatabase: self.unproven = False raise @@ -249,63 +458,57 @@ def reset(self, snapshot: Path, *, use_list: Path | None = None) -> None: restore(self.dsn, snapshot, jobs=self.restore_jobs, use_list=use_list) def close(self) -> None: + # Cancellation is delivered after cleanup, never in the middle of it. + with current().protected(): + self._close() + + def _close(self) -> None: if self.created: - try: - with psycopg.connect(self.admin_dsn, autocommit=True, connect_timeout=10) as conn: - conn.execute("SET statement_timeout = '600s'") - conn.execute("SELECT pg_advisory_lock(%s)", (self.lock_key,)) - row = conn.execute( - "SELECT oid, datdba, shobj_description(oid, 'pg_database') " - "FROM pg_database WHERE datname = %s", - (self.name,), - ).fetchone() - if row is None or (row[0], row[1], row[2]) != ( - self.oid, - self.owner, - self.marker, - ): - raise RuntimeError( - f"Ownership of workspace database {self.name} cannot be verified; " - "inspect it manually before removal" - ) - with self.performance.measure("database_cleanup"): - conn.execute( - sql.SQL("DROP DATABASE {} WITH (FORCE)").format( - sql.Identifier(self.name) - ) - ) - except RuntimeError: - self.cleanup_failures += 1 - raise - except psycopg.Error as error: + assert self.oid is not None and self.owner is not None + result = drop_owned_database( + self.admin_dsn, + self.name, + oid=self.oid, + owner=self.owner, + marker=self.marker, + force=True, + lock_key=self.lock_key, + performance=self.performance, + ) + self.last_cleanup = result + if not result.done: self.cleanup_failures += 1 + if result.status is CleanupStatus.UNVERIFIED: + raise RuntimeError( + f"Ownership of workspace database {self.name} cannot be verified; " + "inspect it manually before removal" + ) raise RuntimeError( - f"Could not confirm cleanup of workspace database {self.name}; " - "inspect and remove it if present" - ) from error + f"Could not confirm cleanup of workspace database {self.name} " + f"({result.detail}); inspect it before any manual removal" + ) from result.error self.created = False self.oid = self.owner = None - self.databases_dropped += 1 + current().record_cleanup(result.status.value) + if result.status is CleanupStatus.REMOVED: + self.databases_dropped += 1 + else: + self.databases_absent += 1 if self.unproven: - try: - with psycopg.connect(self.admin_dsn, connect_timeout=10) as conn: - row = conn.execute( - "SELECT EXISTS (SELECT 1 FROM pg_database WHERE datname = %s)", - (self.name,), - ).fetchone() - except psycopg.Error as error: - self.cleanup_failures += 1 + exists = settle_unproven_create(self.admin_dsn, self.name) + if exists is False: + self.unproven = False + return + self.cleanup_failures += 1 + if exists is None: raise RuntimeError( f"Could not resolve creation outcome for workspace database {self.name}; " "inspect it manually before removal" - ) from error - if row is None or row[0]: - self.cleanup_failures += 1 - raise RuntimeError( - f"Creation outcome for workspace database {self.name} is unverified; " - "inspect it manually before removal" ) - self.unproven = False + raise RuntimeError( + f"Creation outcome for workspace database {self.name} is unverified; " + "inspect it manually before removal" + ) def __exit__( self, diff --git a/src/dbreduce/postgres/dump.py b/src/dbreduce/postgres/dump.py index 1775db9..eaa7890 100644 --- a/src/dbreduce/postgres/dump.py +++ b/src/dbreduce/postgres/dump.py @@ -4,6 +4,8 @@ from psycopg.conninfo import conninfo_to_dict, make_conninfo +from dbreduce.processes import run_owned + class ClientError(RuntimeError): def __init__(self, command: str, reason: str, returncode: int) -> None: @@ -25,14 +27,12 @@ def client(command: list[str], dsn: str) -> None: if password is not None: env["PGPASSWORD"] = str(password) try: - result = subprocess.run( + result = run_owned( [*command, "--dbname", make_conninfo("", **params)], env=env, stdin=subprocess.DEVNULL, - capture_output=True, text=True, timeout=600, - check=False, ) except subprocess.TimeoutExpired as error: raise RuntimeError(f"{command[0]} timed out after 600 seconds") from error diff --git a/src/dbreduce/postgres/schema_reduction.py b/src/dbreduce/postgres/schema_reduction.py index dcac6e8..47309ac 100644 --- a/src/dbreduce/postgres/schema_reduction.py +++ b/src/dbreduce/postgres/schema_reduction.py @@ -18,6 +18,7 @@ from dbreduce.postgres.database import Workspace from dbreduce.postgres.dump import ClientError, dump from dbreduce.postgres.introspection import inspect_schema_counts +from dbreduce.processes import run_owned from dbreduce.reducer.ddmin import chunks, sizes _KINDS = tuple( @@ -86,12 +87,10 @@ class SchemaObject: def archive_objects(path: Path) -> tuple[bytes, list[SchemaObject]]: try: - result = subprocess.run( + result = run_owned( ["pg_restore", "--list", str(path)], - capture_output=True, timeout=600, env={**os.environ, "LC_ALL": "C"}, - check=False, ) except subprocess.TimeoutExpired as error: raise RuntimeError("Timed out reading archive object list") from error diff --git a/src/dbreduce/processes.py b/src/dbreduce/processes.py new file mode 100644 index 0000000..ccfea60 --- /dev/null +++ b/src/dbreduce/processes.py @@ -0,0 +1,98 @@ +"""Subprocesses owned by this run: own process group, bounded stop, no pipe hangs.""" + +import os +import signal +import subprocess +from collections.abc import Mapping, Sequence +from typing import IO, Any + +from dbreduce.cancellation import TERMINATE_GRACE, current + +# After SIGKILL the leader is reaped almost immediately; this only bounds a stuck kernel wait. +REAP_TIMEOUT = 2.0 +# Windows has no SIGKILL; terminating is its only forceful stop. +KILL = getattr(signal, "SIGKILL", signal.SIGTERM) + + +def _signal_group(pgid: int, number: signal.Signals) -> None: + try: + if hasattr(os, "killpg"): + os.killpg(pgid, number) + else: # pragma: no cover - non-POSIX + os.kill(pgid, number) + except OSError: + pass # Already gone. + + +def stop_group(process: "subprocess.Popen[Any]", *, grace: float = TERMINATE_GRACE) -> None: + """Stop the process group started for ``process`` and reap its leader. + + Only groups created by ``start_new_session=True`` for this run are signalled. SIGTERM + is sent first and the leader gets ``grace`` seconds (shortened by the shutdown budget); + then the whole group is killed so no member keeps a pipe or a database session open. + """ + pgid = process.pid + if grace > 0: + wait = current().bounded(grace) + _signal_group(pgid, signal.SIGTERM) + try: + process.wait(timeout=wait) + except subprocess.TimeoutExpired: + pass + _signal_group(pgid, KILL) + try: + process.wait(timeout=REAP_TIMEOUT) + except subprocess.TimeoutExpired: + pass + + +def close_pipes(process: "subprocess.Popen[Any]") -> None: + for stream in (process.stdin, process.stdout, process.stderr): + if stream is not None: + try: + stream.close() + except OSError: + pass + + +def run_owned( + args: Sequence[str], + *, + env: Mapping[str, str] | None = None, + stdin: int | IO[Any] | None = None, + input: str | bytes | None = None, + text: bool = False, + timeout: float | None = None, + check: bool = False, +) -> "subprocess.CompletedProcess[Any]": + """Like ``subprocess.run(..., capture_output=True)`` for a process this run owns. + + On any interruption (timeout, cancellation) the whole process group is stopped and the + pipes are closed without being read again, so a grandchild holding them cannot block. + """ + control = current() + control.checkpoint() + process = subprocess.Popen( + list(args), + env=None if env is None else dict(env), + stdin=subprocess.PIPE if input is not None else stdin, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=text, + start_new_session=True, + ) + control.track_group(process.pid) + try: + stdout, stderr = process.communicate(input=input, timeout=timeout) + except BaseException as error: + # A timeout keeps the old immediate-kill behavior; cancellation gets a graceful stop. + timed_out = isinstance(error, subprocess.TimeoutExpired) + stop_group(process, grace=0 if timed_out else TERMINATE_GRACE) + close_pipes(process) + raise + finally: + control.untrack_group(process.pid) + result = subprocess.CompletedProcess(list(args), process.returncode, stdout, stderr) + if check: + result.check_returncode() + return result diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..e7649cf --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,26 @@ +import os +from functools import lru_cache + +import psycopg +import pytest + + +@lru_cache(maxsize=1) +def server_major(admin: str) -> int: + with psycopg.connect(admin, connect_timeout=10) as conn: + row = conn.execute("SELECT current_setting('server_version_num')::int / 10000").fetchone() + assert row is not None + return int(row[0]) + + +@pytest.fixture(autouse=True) +def _clone_backend_needs_postgres_15(request): + """The clone backend relies on CREATE DATABASE ... STRATEGY, new in PostgreSQL 15.""" + admin = os.environ.get("DBREDUCE_TEST_ADMIN") + if ( + admin + and request.node.get_closest_marker("postgres") + and "clone" in request.node.name + and server_major(admin) < 15 + ): + pytest.skip("The clone backend requires PostgreSQL 15 or later") diff --git a/tests/test_cancellation.py b/tests/test_cancellation.py new file mode 100644 index 0000000..5fe4060 --- /dev/null +++ b/tests/test_cancellation.py @@ -0,0 +1,355 @@ +"""Cancellation without PostgreSQL: signals, shutdown budget, owned subprocesses, oracle.""" + +import os +import shlex +import signal +import subprocess +import sys +import textwrap +import threading +import time +from pathlib import Path +from unittest.mock import patch + +import pytest + +from dbreduce.cancellation import Cancelled, Leftover, RunControl, current +from dbreduce.oracle.runner import Oracle +from dbreduce.processes import run_owned + +posix_only = pytest.mark.skipif(os.name != "posix", reason="POSIX signals and process groups") + + +def alive(pid: int) -> bool: + """True while the process exists and is not a zombie.""" + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + try: # the process may vanish between the checks + return Path(f"/proc/{pid}/stat").read_text().rsplit(")", 1)[1].split()[0] != "Z" + except (FileNotFoundError, ProcessLookupError): + return not Path("/proc/self/stat").exists() # no /proc at all: kill(0) already said alive + + +def wait_for(predicate, timeout=15.0): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + value = predicate() + if value: + return value + time.sleep(0.02) + raise AssertionError("condition not reached in time") + + +def send_to_main_thread_when(path: Path, signum: int = signal.SIGTERM) -> threading.Thread: + """Deliver ``signum`` to the main thread once ``path`` exists (an observed stage).""" + + def run() -> None: + wait_for(path.exists) + signal.pthread_kill(threading.main_thread().ident, signum) + + thread = threading.Thread(target=run, daemon=True) + thread.start() + return thread + + +@pytest.fixture +def control(): + run = RunControl(budget=30.0) + run.start() + try: + yield run + finally: + run.stop() + + +@posix_only +def test_import_installs_no_signal_handlers(): + code = ( + "import signal, dbreduce.cli, dbreduce.cancellation;" + "print(signal.getsignal(signal.SIGTERM) is signal.SIG_DFL," + " signal.getsignal(signal.SIGINT) is signal.default_int_handler)" + ) + result = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True) + assert result.stdout.split() == ["True", "True"], result.stderr + + +@posix_only +def test_handlers_are_installed_for_a_run_and_restored(): + before = {n: signal.getsignal(n) for n in (signal.SIGINT, signal.SIGTERM)} + run = RunControl() + run.start() + try: + assert signal.getsignal(signal.SIGTERM) != before[signal.SIGTERM] + assert signal.getsignal(signal.SIGINT) != before[signal.SIGINT] + finally: + run.stop() + assert {n: signal.getsignal(n) for n in before} == before + assert current() is not run + + +@posix_only +def test_handlers_are_not_installed_outside_the_main_thread(): + before = signal.getsignal(signal.SIGTERM) + seen = [] + + def start_and_stop(): + run = RunControl() + run.start() + seen.append(signal.getsignal(signal.SIGTERM)) + run.stop() + + thread = threading.Thread(target=start_and_stop) + thread.start() + thread.join() + assert seen == [before] + + +@posix_only +def test_signal_raises_cancelled_once_and_repeats_are_ignored(control): + with pytest.raises(Cancelled) as raised: + os.kill(os.getpid(), signal.SIGTERM) + time.sleep(5) + assert raised.value.signum == signal.SIGTERM + assert isinstance(raised.value, KeyboardInterrupt) + assert control.signal_name == "SIGTERM" + assert control.remaining() is not None and 0 < control.remaining() <= 30.0 + # A second signal while cleanup runs is recorded, not raised. + os.kill(os.getpid(), signal.SIGINT) + time.sleep(0.2) + assert control.ignored_signals == 1 + + +@posix_only +def test_sigint_cancels_like_sigterm(control): + with pytest.raises(Cancelled) as raised: + os.kill(os.getpid(), signal.SIGINT) + time.sleep(5) + assert raised.value.signum == signal.SIGINT + + +@posix_only +def test_protected_section_defers_cancellation_until_it_ends(control): + steps = [] + with pytest.raises(Cancelled): + with control.protected(): + os.kill(os.getpid(), signal.SIGTERM) + time.sleep(0.2) + steps.append("section finished") + steps.append("not reached") + assert steps == ["section finished"] + # Once delivered, later protected sections (cleanup) are not interrupted again. + with control.protected(): + os.kill(os.getpid(), signal.SIGTERM) + time.sleep(0.2) + assert control.ignored_signals == 1 + + +@posix_only +def test_no_new_work_after_cancellation(control): + with pytest.raises(Cancelled): # delivered when the protected section ends + with control.protected(): + os.kill(os.getpid(), signal.SIGTERM) + time.sleep(0.1) + with pytest.raises(Cancelled): + control.checkpoint() + with pytest.raises(Cancelled): + run_owned([sys.executable, "-c", "pass"]) + + +@posix_only +def test_signal_after_commit_does_not_cancel(control): + control.commit() + os.kill(os.getpid(), signal.SIGTERM) + time.sleep(0.2) + assert control.cancelled and control.committed + + +def test_budget_shrinks_timeouts_and_expires(): + run = RunControl(budget=100.0) + assert run.bounded(600) == 600 and not run.expired() + run.requested = signal.SIGTERM + run.deadline = time.monotonic() + 3 + assert 1.0 <= run.bounded(600) <= 3.0 + run.deadline = time.monotonic() - 1 + assert run.expired() and run.bounded(600) == 1.0 + + +@posix_only +def test_watchdog_forces_exit_when_cleanup_never_returns(tmp_path): + script = textwrap.dedent( + """ + import subprocess, sys, time + import dbreduce.cancellation as c + from dbreduce.cancellation import Leftover, RunControl + c.REPORT_MARGIN = 0.5 + run = RunControl(budget=1.0) + run.start() + run.register(lambda: [Leftover("database", "dbreduce_example", "removal not attempted")]) + owned = subprocess.Popen(["sleep", "60"], start_new_session=True) + run.track_group(owned.pid) + print(owned.pid, flush=True) + with run.protected(): # cleanup that blocks forever + time.sleep(120) + """ + ) + process = subprocess.Popen( + [sys.executable, "-c", script], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + owned_pid = int(process.stdout.readline()) + started = time.monotonic() + process.send_signal(signal.SIGTERM) + _, err = process.communicate(timeout=15) + assert time.monotonic() - started < 6 + assert process.returncode == 1 + assert "forcing exit" in err and "NOT confirmed" in err + assert "dbreduce_example" in err + wait_for(lambda: not alive(owned_pid), timeout=5) + + +@posix_only +def test_run_owned_timeout_kills_group_without_waiting_for_pipes(tmp_path): + pidfile = tmp_path / "grandchild.pid" + command = f"sleep 60 & echo $! > {shlex.quote(str(pidfile))}; wait" + started = time.monotonic() + with pytest.raises(subprocess.TimeoutExpired): + run_owned(["sh", "-c", command], timeout=1.0) + assert time.monotonic() - started < 8 + grandchild = int(pidfile.read_text()) + wait_for(lambda: not alive(grandchild), timeout=5) + + +@posix_only +@pytest.mark.parametrize("cooperative", [True, False]) +def test_run_owned_cancellation_terminates_then_kills(control, tmp_path, cooperative): + flag = tmp_path / "started" + seen = tmp_path / "term-seen" + script = textwrap.dedent( + f""" + import os, signal, sys, time + def on_term(*_): + open({str(seen)!r}, "w").close() + sys.exit(0) + signal.signal(signal.SIGTERM, on_term if {cooperative} else signal.SIG_IGN) + open({str(flag)!r}, "w").write(str(os.getpid())) + time.sleep(120) + """ + ) + trigger = send_to_main_thread_when(flag) + started = time.monotonic() + with patch("dbreduce.processes.TERMINATE_GRACE", 1.0): + with pytest.raises(Cancelled): + run_owned([sys.executable, "-c", script], timeout=120) + trigger.join() + assert time.monotonic() - started < 10 + assert not alive(int(flag.read_text())) + # The graceful SIGTERM reached a cooperative process before any SIGKILL. + assert seen.exists() is cooperative + + +@posix_only +@pytest.mark.parametrize("cooperative", [True, False]) +def test_oracle_cancellation_stops_its_process_group(control, tmp_path, cooperative): + flag = tmp_path / "started" + seen = tmp_path / "term-seen" + child_pid = tmp_path / "child.pid" + script = textwrap.dedent( + f""" + import os, signal, subprocess, sys, time + def on_term(*_): + open({str(seen)!r}, "w").close() + sys.exit(0) + signal.signal(signal.SIGTERM, on_term if {cooperative} else signal.SIG_IGN) + child = subprocess.Popen(["sleep", "120"]) # a grandchild of the supervisor + open({str(child_pid)!r}, "w").write(str(child.pid)) + open({str(flag)!r}, "w").write(str(os.getpid())) + time.sleep(120) + """ + ) + # exec: the command itself is in the group, not a shell that would die on SIGTERM. + ignore = "" if cooperative else "trap '' TERM; " + oracle = Oracle( + f"{ignore}exec {shlex.quote(sys.executable)} -c {shlex.quote(script)}", timeout=120 + ) + trigger = send_to_main_thread_when(flag) + started = time.monotonic() + with patch("dbreduce.oracle.runner.TERMINATE_GRACE", 1.0): + with pytest.raises(Cancelled): + oracle.fails("postgresql:///test", lambda: None) + trigger.join() + assert time.monotonic() - started < 10 + assert not alive(int(flag.read_text())) + wait_for(lambda: not alive(int(child_pid.read_text())), timeout=5) + assert seen.exists() is cooperative + # Cancellation is not a verdict about the candidate. + assert oracle.executions == 1 and not oracle.outcomes + + +@posix_only +def test_oracle_started_after_cancellation_is_refused(control): + with pytest.raises(Cancelled): + with control.protected(): + os.kill(os.getpid(), signal.SIGTERM) + time.sleep(0.1) + oracle = Oracle(f"{shlex.quote(sys.executable)} -c pass") + with pytest.raises(Cancelled): + oracle.fails("postgresql:///test", lambda: None) + assert oracle.executions == 0 + + +def test_leftover_lines_name_each_resource(): + from dbreduce.cancellation import describe_leftovers + + lines = describe_leftovers([Leftover("database", "dbreduce_x", "result unknown", "why")]) + assert lines == [" database dbreduce_x: result unknown", " why"] + + +@posix_only +def test_inherited_ignore_disposition_is_kept(): + code = ( + "import os, signal, time\n" + "from dbreduce.cancellation import RunControl\n" + "run = RunControl(); run.start()\n" + "kept = signal.getsignal(signal.SIGINT) == signal.SIG_IGN\n" + "os.kill(os.getpid(), signal.SIGINT); time.sleep(0.3)\n" + "print(kept, run.cancelled); run.stop()\n" + ) + result = subprocess.run( + [sys.executable, "-c", code], + capture_output=True, + text=True, + preexec_fn=lambda: signal.signal(signal.SIGINT, signal.SIG_IGN), + ) + assert result.stdout.split() == ["True", "False"], result.stderr + + +@posix_only +def test_watchdog_kills_a_running_oracle_group(tmp_path): + pidfile = tmp_path / "oracle.pid" + script = textwrap.dedent( + f""" + import sys, time + import dbreduce.cancellation as c + from dbreduce.cancellation import RunControl + from dbreduce.oracle.runner import Oracle + c.REPORT_MARGIN = 0.5 + run = RunControl(budget=1.0) + run.start() + oracle = Oracle("echo $$ > {pidfile}; exec sleep 120", timeout=120) + with run.protected(): # cancellation is deferred, so only the watchdog can end this + oracle.fails("postgresql:///test", lambda: None) + """ + ) + process = subprocess.Popen( + [sys.executable, "-c", script], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True + ) + wait_for(pidfile.exists) + oracle_pid = int(wait_for(lambda: pidfile.read_text().strip())) + process.send_signal(signal.SIGTERM) + process.communicate(timeout=15) + assert process.returncode == 1 + wait_for(lambda: not alive(oracle_pid), timeout=5) diff --git a/tests/test_cancellation_postgres.py b/tests/test_cancellation_postgres.py new file mode 100644 index 0000000..0e5c6b5 --- /dev/null +++ b/tests/test_cancellation_postgres.py @@ -0,0 +1,769 @@ +"""Cancellation of a real ``dbreduce reduce`` process against a disposable PostgreSQL. + +Every test starts the CLI in its own process and signals it only after an observed stage: +the oracle (or a shimmed ``pg_restore``) writes a handshake once it is running, or the +server shows the expected session. The tests send signals straight to the DBReduce process +or to its process group; they do not reproduce ``docker stop`` or a CI runner's cancel. + +Test cleanup uses raw SQL on names the test itself recorded or observed during the test, so it +never depends on the DBReduce code under test. +""" + +import hashlib +import json +import os +import re +import shlex +import shutil +import signal +import socket +import subprocess +import sys +import threading +import time +import uuid +from pathlib import Path +from types import SimpleNamespace + +import psycopg +import pytest +from psycopg import sql +from psycopg.conninfo import conninfo_to_dict, make_conninfo + +from dbreduce.postgres.database import database_url + +pytestmark = pytest.mark.postgres + +GENERATED = re.compile(r"^dbreduce_[0-9a-f]{32}$") +EXTERNAL_TIMEOUT = 90 # seconds a DBReduce process may take before the test kills it +IS_LINUX = Path("/proc/self/environ").exists() + +LAUNCHER = r""" +import os, sys, time +from pathlib import Path +import dbreduce.cancellation as cancellation + +if "TEST_BUDGET" in os.environ: + cancellation.SHUTDOWN_BUDGET = float(os.environ["TEST_BUDGET"]) + cancellation.REPORT_MARGIN = float(os.environ.get("TEST_MARGIN", "2")) + +hook = os.environ.get("TEST_HOOK") +if hook: + import psycopg + from psycopg import sql + + original = psycopg.Connection.execute + nth = int(os.environ.get("TEST_HOOK_NTH", "1")) + seen = [0] + + def execute(self, query, *args, **kwargs): + text = query.as_string(self) if isinstance(query, sql.Composable) else str(query) + if text.startswith("CREATE DATABASE"): + seen[0] += 1 + if seen[0] == nth: + name = text.split()[2].strip('"') + Path(os.environ["TEST_STATE"], "hook-database").write_text(name) + if hook == "during_create": + # A statement that is still running on the server when the client stops. + return original(self, "SELECT pg_sleep(3600)") + result = original(self, query, *args, **kwargs) + time.sleep(3600) # the database exists; the run marker was not written yet + return result + return original(self, query, *args, **kwargs) + + psycopg.Connection.execute = execute + +from dbreduce.cli import app +app() +""" + +ORACLE = r""" +import json, os, signal, sys, time +from pathlib import Path +import psycopg + +state = Path(os.environ["TEST_STATE"]) +with (state / "calls").open("a") as calls: + calls.write("x\n") +count = len((state / "calls").read_text().split()) +connection = psycopg.connect(os.environ["DATABASE_URL"]) +if count == int(os.environ["TEST_BLOCK_ON"]): + connection.execute("SELECT 1") # a live session on the database under test + mode = os.environ["TEST_ORACLE_MODE"] + if mode == "term": + def on_term(*_): + (state / "term-seen").write_text("1") + sys.exit(0) + signal.signal(signal.SIGTERM, on_term) + else: + signal.signal(signal.SIGTERM, signal.SIG_IGN) + (state / "blocked.json").write_text( + json.dumps({"pid": os.getpid(), "database": connection.info.dbname}) + ) + while True: + time.sleep(1) +bug = connection.execute("SELECT EXISTS(SELECT 1 FROM child WHERE id = 1)").fetchone()[0] +print(json.dumps({"reproduced": bug, "signature": "BUG_A" if bug else None})) +""" + +PG_RESTORE_SHIM = """#!/bin/sh +# Stands in for pg_restore: a child that keeps the output pipes open outlives its parent. +echo $$ > "$TEST_STATE/restore.pid" +sleep 600 & +echo $! > "$TEST_STATE/restore-child.pid" +echo started > "$TEST_STATE/restore.started" +wait +""" + + +def wait_for(predicate, timeout=60.0, interval=0.05, what="condition"): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + value = predicate() + if value: + return value + time.sleep(interval) + raise AssertionError(f"{what} not reached within {timeout} s") + + +def process_alive(pid: int) -> bool: + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + try: # the process may vanish between the checks + return Path(f"/proc/{pid}/stat").read_text().rsplit(")", 1)[1].split()[0] != "Z" + except (FileNotFoundError, ProcessLookupError): + return not Path("/proc/self/stat").exists() # no /proc at all: kill(0) already said alive + + +def marked_processes(run_id: str) -> list[int]: + """Live processes that inherited this run's environment marker (all descendants).""" + if not IS_LINUX: + return [] + needle = f"TEST_RUN_ID={run_id}".encode() + found = [] + for entry in Path("/proc").iterdir(): + if not entry.name.isdigit(): + continue + try: + if needle in (entry / "environ").read_bytes().split(b"\0") and process_alive( + int(entry.name) + ): + found.append(int(entry.name)) + except OSError: + continue + return found + + +class Server: + def __init__(self, dsn: str) -> None: + self.dsn = dsn + + def connect(self, **kwargs): + return psycopg.connect(self.dsn, autocommit=True, connect_timeout=10, **kwargs) + + def names(self) -> set[str]: + with self.connect() as conn: + return {row[0] for row in conn.execute("SELECT datname FROM pg_database")} + + def generated(self) -> set[str]: + return {name for name in self.names() if GENERATED.match(name)} + + def comment(self, name: str): + with self.connect() as conn: + row = conn.execute( + "SELECT shobj_description(oid, 'pg_database') FROM pg_database WHERE datname = %s", + (name,), + ).fetchone() + return None if row is None else row[0] + + def drop(self, name: str) -> None: + with self.connect() as conn: + conn.execute( + sql.SQL("DROP DATABASE IF EXISTS {} WITH (FORCE)").format(sql.Identifier(name)) + ) + + def dsn_for(self, name: str) -> str: + return make_conninfo(self.dsn, dbname=name) + + +@pytest.fixture +def env(tmp_path): + if not IS_LINUX: + pytest.skip("Process checks read /proc; run these tests on Linux") + admin = os.environ.get("DBREDUCE_TEST_ADMIN") + if not admin or not all(shutil.which(tool) for tool in ("pg_dump", "pg_restore", "psql")): + pytest.skip("Set DBREDUCE_TEST_ADMIN and install PostgreSQL client tools") + server = Server(admin) + suffix = uuid.uuid4().hex[:12] + source, foreign = f"cancel_source_{suffix}", f"cancel_foreign_{suffix}" + lookalike = f"dbreduce_{uuid.uuid4().hex}" # looks generated but belongs to someone else + for name in (source, foreign, lookalike): + with server.connect() as conn: + conn.execute(sql.SQL("CREATE DATABASE {}").format(sql.Identifier(name))) + with psycopg.connect(server.dsn_for(source)) as conn: + conn.execute(""" + CREATE TABLE parent (id integer PRIMARY KEY); + CREATE TABLE child (id integer PRIMARY KEY, parent_id integer REFERENCES parent); + INSERT INTO parent SELECT generate_series(1, 20); + INSERT INTO child SELECT i, i FROM generate_series(1, 20) AS i; + """) + foreign_session = psycopg.connect(server.dsn_for(foreign), autocommit=True) + foreign_session.execute("CREATE TABLE keep (id integer); INSERT INTO keep VALUES (1)") + foreign_pid = foreign_session.info.backend_pid + before = server.generated() + assert lookalike in before + context = SimpleNamespace( + server=server, + source=source, + foreign=foreign, + lookalike=lookalike, + foreign_session=foreign_session, + foreign_pid=foreign_pid, + before=before, + tmp=tmp_path, + fingerprint=lambda: source_fingerprint(server.dsn_for(source)), + ) + context.initial_fingerprint = context.fingerprint() + try: + yield context + finally: + # Independent of the code under test: drop only what appeared while this test ran. + foreign_session.close() + for name in server.generated() - before: + server.drop(name) + for name in (source, foreign, lookalike): + server.drop(name) + + +def source_fingerprint(dsn: str): + with psycopg.connect(dsn) as conn: + return conn.execute(""" + SELECT (SELECT md5(string_agg(id::text, ',' ORDER BY id)) FROM parent), + (SELECT md5(string_agg(id || ':' || parent_id, ',' ORDER BY id)) FROM child), + (SELECT oid FROM pg_database WHERE datname = current_database()) + """).fetchone() + + +class Run: + """One DBReduce process started for a test.""" + + def __init__( + self, + env, + *, + dsn=None, + backend="snapshot", + block_on=2, + oracle_mode="term", + hook=None, + hook_nth=1, + budget=None, + margin=None, + path_prefix=None, + ): + self.env = env + self.state = env.tmp / "state" + self.state.mkdir() + self.run_id = uuid.uuid4().hex + self.cli_tmp = env.tmp / "cli-tmp" + self.cli_tmp.mkdir() + self.output, self.report = env.tmp / "min.sql", env.tmp / "report.json" + (env.tmp / "oracle.py").write_text(ORACLE) + admin = dsn or env.server.dsn + source = make_conninfo(admin, dbname=env.source) + process_env = { + **os.environ, + "TMPDIR": str(self.cli_tmp), + "TEST_STATE": str(self.state), + "TEST_RUN_ID": self.run_id, + "TEST_BLOCK_ON": str(block_on), + "TEST_ORACLE_MODE": oracle_mode, + "PYTHONDONTWRITEBYTECODE": "1", + } + if budget is not None: + process_env["TEST_BUDGET"] = str(budget) + process_env["TEST_MARGIN"] = str(margin if margin is not None else 2) + if hook: + process_env["TEST_HOOK"] = hook + process_env["TEST_HOOK_NTH"] = str(hook_nth) + if path_prefix: + process_env["PATH"] = f"{path_prefix}{os.pathsep}{process_env['PATH']}" + # exec: the oracle itself is the process in the group, not a shell that dies on SIGTERM. + ignore = "trap '' TERM; " if oracle_mode == "ignore" else "" + oracle = ( + f"{ignore}exec {shlex.quote(sys.executable)} {shlex.quote(str(env.tmp / 'oracle.py'))}" + ) + command = [ + sys.executable, "-c", LAUNCHER, "reduce", + "--database", source, "--admin-database", admin, + "--oracle", oracle, "--oracle-json", "--timeout", "120", + "--candidate-backend", backend, + "--output", str(self.output), "--report", str(self.report), + ] # fmt: skip + self.started = time.monotonic() + self.process = subprocess.Popen( + command, env=process_env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + text=True, start_new_session=True, + ) # fmt: skip + self.stdout = self.stderr = "" + + def wait_for_file(self, name: str, what: str) -> Path: + path = self.state / name + + def reached(): + if self.process.poll() is not None: + out, err = self.process.communicate() + raise AssertionError(f"DBReduce exited before {what}:\n{out}\n{err}") + return path.exists() + + wait_for(reached, timeout=EXTERNAL_TIMEOUT, what=what) + return path + + def blocked(self) -> dict: + self.wait_for_file("blocked.json", "the oracle was running") + return json.loads((self.state / "blocked.json").read_text()) + + def send(self, signum, *, group=False): + if group: + os.killpg(self.process.pid, signum) + else: + os.kill(self.process.pid, signum) + + def finish(self, timeout=EXTERNAL_TIMEOUT) -> int: + try: + self.stdout, self.stderr = self.process.communicate(timeout=timeout) + except subprocess.TimeoutExpired: + self.kill_everything() + self.stdout, self.stderr = self.process.communicate() + pytest.fail(f"DBReduce did not exit within {timeout} s:\n{self.stderr}") + self.elapsed = time.monotonic() - self.started + return self.process.returncode + + def kill_everything(self): + """Test-side cleanup of the process tree, independent of DBReduce's own.""" + for pid in marked_processes(self.run_id): + try: + os.kill(pid, signal.SIGKILL) + except ProcessLookupError: + pass + try: + os.killpg(self.process.pid, signal.SIGKILL) + except ProcessLookupError: + pass + + def oracle_calls(self) -> int: + path = self.state / "calls" + return len(path.read_text().split()) if path.exists() else 0 + + +@pytest.fixture +def start(env): + runs: list[Run] = [] + + def factory(**kwargs) -> Run: + run = Run(env, **kwargs) + runs.append(run) + return run + + yield factory + for run in runs: + if run.process.poll() is None: + run.kill_everything() + run.process.communicate() + else: + run.kill_everything() + + +def assert_isolated(env, run: Run, *, expect_dirs=True): + """The source, the foreign database and its live session are untouched; nothing published.""" + assert env.fingerprint() == env.initial_fingerprint + assert env.lookalike in env.server.names() + assert env.server.comment(env.lookalike) is None + assert env.foreign_session.execute("SELECT count(*) FROM keep").fetchone()[0] == 1 + with env.server.connect() as conn: + assert ( + conn.execute( + "SELECT count(*) FROM pg_stat_activity WHERE pid = %s AND datname = %s", + (env.foreign_pid, env.foreign), + ).fetchone()[0] + == 1 + ) + assert not run.output.exists() and not run.report.exists() + assert not list(run.env.tmp.glob(".dbreduce-*")) + if expect_dirs: + assert list(run.cli_tmp.iterdir()) == [] + + +def assert_no_owned_leftovers(env, run: Run): + assert env.server.generated() - env.before == set() + assert marked_processes(run.run_id) == [] + + +# -- SIGTERM and SIGINT while an oracle is running -------------------------------------- + + +@pytest.mark.parametrize( + ("backend", "block_on", "oracle_mode", "target"), + [ + ("snapshot", 2, "term", "process"), + ("snapshot", 2, "term", "group"), + ("snapshot", 2, "ignore", "process"), + ("clone", 3, "term", "process"), + ], +) +def test_sigterm_during_active_oracle_cleans_everything_owned( + env, start, backend, block_on, oracle_mode, target +): + run = start(backend=backend, block_on=block_on, oracle_mode=oracle_mode) + info = run.blocked() + owned_now = env.server.generated() - env.before + assert len(owned_now) >= (4 if backend == "clone" else 1) + assert info["database"] in owned_now + oracle_pid = info["pid"] + run.send(signal.SIGTERM, group=target == "group") + code = run.finish() + assert code == 128 + signal.SIGTERM, run.stderr + assert "Cancelled by SIGTERM" in run.stderr + assert "Cleanup complete" in run.stderr + assert "NOT complete" not in run.stderr + assert run.elapsed < 30 + assert run.oracle_calls() == block_on # no oracle call started after the signal + assert not process_alive(oracle_pid) + assert (run.state / "term-seen").exists() is (oracle_mode == "term") + assert_no_owned_leftovers(env, run) + assert_isolated(env, run) + + +def test_sigint_keeps_expected_behavior(env, start): + run = start() + run.blocked() + run.send(signal.SIGINT) + code = run.finish() + assert code == 128 + signal.SIGINT, run.stderr + assert "Cancelled by SIGINT" in run.stderr and "Cleanup complete" in run.stderr + assert_no_owned_leftovers(env, run) + assert_isolated(env, run) + + +def test_unfinished_create_of_this_run_is_cancelled_and_other_sessions_are_left_alone(env): + """The client vanishes without cancelling; only the session named after the database stops.""" + from dbreduce.postgres.database import settle_unproven_create + + name = f"dbreduce_{uuid.uuid4().hex}" + mine = psycopg.connect(env.server.dsn, autocommit=True, application_name=name) + other = psycopg.connect(env.server.dsn, autocommit=True, application_name="someone_else") + other_pid = other.info.backend_pid + + def hold(connection): + try: + connection.execute("SELECT pg_sleep(3600)") + except psycopg.Error: + pass + + threads = [threading.Thread(target=hold, args=(c,), daemon=True) for c in (mine, other)] + for thread in threads: + thread.start() + + def active(application): + with env.server.connect() as conn: + return conn.execute( + "SELECT count(*) FROM pg_stat_activity " + "WHERE application_name = %s AND state = 'active'", + (application,), + ).fetchone()[0] + + try: + wait_for(lambda: active(name) and active("someone_else"), what="both sessions running") + with socket.socket(fileno=os.dup(mine.pgconn.socket)) as connection_socket: + connection_socket.shutdown(socket.SHUT_RDWR) # lost connection, no cancel request + assert settle_unproven_create(env.server.dsn, name) is False + assert active(name) == 0 + assert active("someone_else") == 1 + finally: + with env.server.connect() as conn: + conn.execute("SELECT pg_terminate_backend(%s)", (other_pid,)) + for thread in threads: + thread.join(timeout=30) + mine.close() + other.close() + + +# -- stages that need a controlled hold --------------------------------------------------- + + +def test_sigterm_while_pg_restore_and_its_child_hold_the_pipes(env, start): + bin_dir = env.tmp / "bin" + bin_dir.mkdir() + shim = bin_dir / "pg_restore" + shim.write_text(PG_RESTORE_SHIM) + shim.chmod(0o755) + run = start(path_prefix=str(bin_dir)) + run.wait_for_file("restore.started", "pg_restore was running") + assert len(env.server.generated() - env.before) == 1 # the workspace, mid-restore + run.send(signal.SIGTERM) + code = run.finish() + assert code == 128 + signal.SIGTERM, run.stderr + assert "Cleanup complete" in run.stderr + assert run.elapsed < 30 # not held back by the child that kept the pipes open + for name in ("restore.pid", "restore-child.pid"): + assert not process_alive(int((run.state / name).read_text())) + assert_no_owned_leftovers(env, run) + assert_isolated(env, run) + + +@pytest.mark.parametrize("backend", ["snapshot", "clone"]) +def test_cancel_after_create_before_ownership_is_recorded_reports_uncertainty(env, start, backend): + run = start(backend=backend, hook="after_create", hook_nth=1 if backend == "snapshot" else 2) + created = run.wait_for_file("hook-database", "CREATE DATABASE returned").read_text() + # The hook sleeps right after the server created the database, before the marker. + wait_for(lambda: created in env.server.names(), what="the database appearing") + assert env.server.comment(created) is None + run.send(signal.SIGTERM) + code = run.finish() + try: + assert code == 1, run.stderr + assert "Cancelled by SIGTERM" in run.stderr + assert "Cleanup is NOT complete" in run.stderr + assert created in run.stderr and "creation outcome unverified" in run.stderr + # A name alone is not ownership proof: the database was reported, not dropped. + assert created in env.server.names() + assert env.server.comment(created) is None + # Everything that did carry this run's marker was removed. + unmarked = env.server.generated() - env.before - {created} + assert unmarked == set() + assert marked_processes(run.run_id) == [] + assert_isolated(env, run) + finally: + env.server.drop(created) + + +def test_cancel_during_create_stops_the_statement_still_running_on_the_server(env, start): + run = start(hook="during_create") + name = run.wait_for_file("hook-database", "CREATE DATABASE was issued").read_text() + + def statement_running(): + with env.server.connect() as conn: + return conn.execute( + "SELECT count(*) FROM pg_stat_activity WHERE application_name = %s " + "AND state = 'active' AND query LIKE 'SELECT pg_sleep%%'", + (name,), + ).fetchone()[0] + + wait_for(statement_running, what="the statement running on the server") + run.send(signal.SIGTERM) + code = run.finish() + assert code == 128 + signal.SIGTERM, run.stderr + assert "Cleanup complete" in run.stderr + with env.server.connect() as conn: # the server-side statement did not outlive the run + assert ( + conn.execute( + "SELECT count(*) FROM pg_stat_activity WHERE application_name = %s", (name,) + ).fetchone()[0] + == 0 + ) + assert name not in env.server.names() + assert_no_owned_leftovers(env, run) + assert_isolated(env, run) + + +# -- PostgreSQL unavailable or busy during cleanup -------------------------------------- + + +class Proxy: + """A TCP forwarder the test can cut, to make the server unreachable on demand.""" + + def __init__(self, host: str, port: int) -> None: + self.target = (host, port) + self.listener = socket.socket() + self.listener.bind(("127.0.0.1", 0)) + self.listener.listen() + self.port = self.listener.getsockname()[1] + self.sockets: list[socket.socket] = [] + threading.Thread(target=self._accept, daemon=True).start() + + def _accept(self): + while True: + try: + client, _ = self.listener.accept() + upstream = socket.create_connection(self.target) + except OSError: + return + self.sockets += [client, upstream] + for source, sink in ((client, upstream), (upstream, client)): + threading.Thread(target=self._pipe, args=(source, sink), daemon=True).start() + + @staticmethod + def _pipe(source, sink): + try: + while data := source.recv(65536): + sink.sendall(data) + except OSError: + pass + finally: + for s in (source, sink): + try: + s.shutdown(socket.SHUT_RDWR) + except OSError: + pass + + def stop(self): + try: + self.listener.shutdown(socket.SHUT_RDWR) # wakes accept(); close() alone does not + except OSError: + pass + self.listener.close() + for s in self.sockets: + try: + s.shutdown(socket.SHUT_RDWR) + s.close() + except OSError: + pass + + +def test_unreachable_server_during_cleanup_is_finite_and_reported(env, start): + params = conninfo_to_dict(env.server.dsn) + proxy = Proxy(str(params["host"]), int(params.get("port", 5432))) + try: + via_proxy = make_conninfo(env.server.dsn, host="127.0.0.1", port=proxy.port) + run = start(dsn=via_proxy) + info = run.blocked() + workspace = info["database"] + proxy.stop() # the server disappears while the oracle runs + run.send(signal.SIGTERM) + code = run.finish() + assert code == 1, run.stderr + assert run.elapsed < 40 + assert "Cancelled by SIGTERM" in run.stderr + assert "Cleanup is NOT complete" in run.stderr + assert workspace in run.stderr and "result unknown" in run.stderr + assert "temporary directory removed" in run.stderr # local cleanup is reported apart + assert "Cleanup complete" not in run.stderr + # The report gives the way to check ownership; the database is still there, marked. + assert "shobj_description" in run.stderr and "dbreduce-workspace:" in run.stderr + assert env.server.comment(workspace).startswith("dbreduce-workspace:") + assert marked_processes(run.run_id) == [] + assert_isolated(env, run) + finally: + proxy.stop() + # What the report tells an operator to do: verify the comment, then drop by exact name. + for name in env.server.generated() - env.before: + if str(env.server.comment(name) or "").startswith("dbreduce-workspace:"): + env.server.drop(name) + + +def advisory_key(name: str) -> int: + return int.from_bytes(hashlib.sha256(name.encode()).digest()[:8], "big", signed=True) + + +def waiting_for_advisory_lock(env) -> bool: + with env.server.connect() as conn: + return bool( + conn.execute( + "SELECT count(*) FROM pg_locks WHERE locktype = 'advisory' AND NOT granted" + ).fetchone()[0] + ) + + +def test_repeated_signals_do_not_interrupt_cleanup(env, start): + run = start() + workspace = run.blocked()["database"] + holder = env.server.connect() + holder.execute("SELECT pg_advisory_lock(%s)", (advisory_key(workspace),)) + try: + run.send(signal.SIGTERM) # cleanup now waits for the workspace lock + wait_for(lambda: waiting_for_advisory_lock(env), what="cleanup waiting for the lock") + run.send(signal.SIGTERM) + run.send(signal.SIGINT) + time.sleep(1) + assert run.process.poll() is None # extra signals did not abort the cleanup + holder.execute("SELECT pg_advisory_unlock(%s)", (advisory_key(workspace),)) + code = run.finish() + finally: + holder.close() + assert code == 128 + signal.SIGTERM, run.stderr + assert "again" in run.stderr and "SIGQUIT" in run.stderr + assert "Cleanup complete" in run.stderr + assert_no_owned_leftovers(env, run) + assert_isolated(env, run) + + +def test_sigquit_abandons_a_stuck_cleanup(env, start): + run = start() + workspace = run.blocked()["database"] + holder = env.server.connect() + holder.execute("SELECT pg_advisory_lock(%s)", (advisory_key(workspace),)) + try: + run.send(signal.SIGTERM) + wait_for(lambda: waiting_for_advisory_lock(env), what="cleanup waiting for the lock") + run.send(signal.SIGQUIT) + code = run.finish(timeout=20) + finally: + holder.close() + run.kill_everything() + assert code == -signal.SIGQUIT + # Abandoning cleanup is documented to leave the workspace; the test removes it. + assert workspace in env.server.names() + + +def test_cleanup_wait_for_a_lock_is_bounded_by_the_shutdown_budget(env, start): + run = start(budget=8, margin=6) + workspace = run.blocked()["database"] + holder = env.server.connect() + holder.execute("SELECT pg_advisory_lock(%s)", (advisory_key(workspace),)) + try: + run.send(signal.SIGTERM) + code = run.finish(timeout=40) + finally: + holder.close() + assert code == 1, run.stderr + assert 7 <= run.elapsed < 8 + 6, run.elapsed + assert "forcing exit" not in run.stderr # cooperative cleanup gave up on its own + assert "Cleanup is NOT complete" in run.stderr + assert workspace in run.stderr and "not removed" in run.stderr + assert env.server.comment(workspace).startswith("dbreduce-workspace:") + assert marked_processes(run.run_id) == [] + assert_isolated(env, run) + + +# -- ordinary reduction is unchanged ------------------------------------------------------ + + +@pytest.mark.parametrize("backend", ["snapshot", "clone"]) +def test_successful_reduction_keeps_failure_identity_and_export_reproduces(env, start, backend): + run = start(backend=backend, block_on=0) + code = run.finish() + assert code == 0, run.stderr + report = json.loads(run.report.read_text()) + assert report["final_rows"] == 2 + assert report["failure_identity"]["preserved"] is True + expected = "sha256:" + hashlib.sha256(b"BUG_A").hexdigest() + assert report["failure_identity"]["expected_signature"] == expected + assert report["database_stats"]["cleanup_failures"] == 0 + assert_no_owned_leftovers(env, run) + assert env.fingerprint() == env.initial_fingerprint + # A fresh restore of the exported SQL still shows the same failure. + restored = report["restore_database"] + subprocess.run( + ["psql", "-X", "-q", "-v", "ON_ERROR_STOP=1", "-d", env.server.dsn, "-f", str(run.output)], + check=True, + capture_output=True, + ) + try: + verdict = subprocess.run( + [sys.executable, str(env.tmp / "oracle.py")], + env={ + **os.environ, + "DATABASE_URL": database_url(env.server.dsn, restored), + "TEST_STATE": str(run.state), + "TEST_BLOCK_ON": "0", + "TEST_ORACLE_MODE": "term", + }, + check=True, + capture_output=True, + text=True, + ) + assert json.loads(verdict.stdout) == {"reproduced": True, "signature": "BUG_A"} + finally: + env.server.drop(restored) diff --git a/tests/test_cli.py b/tests/test_cli.py index 073d736..cfbb256 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -106,6 +106,7 @@ def test_reduce_json_identity_report(tmp_path, monkeypatch, final_invalid): workspace.url = "postgresql:///dbreduce_test" workspace.databases_created = 0 workspace.databases_dropped = 0 + workspace.databases_absent = 0 workspace.cleanup_failures = 0 from collections import Counter diff --git a/tests/test_postgres.py b/tests/test_postgres.py index bf54f4b..dfb846c 100644 --- a/tests/test_postgres.py +++ b/tests/test_postgres.py @@ -179,6 +179,8 @@ def test_source_encoding_and_locale_are_copied(provider): pytest.skip("Set DBREDUCE_TEST_ADMIN") if provider == "i": with psycopg.connect(admin) as conn: + if conn.info.server_version < 150000: + pytest.skip("Per-database ICU locale providers require PostgreSQL 15") if not conn.execute( "SELECT EXISTS (SELECT 1 FROM pg_collation WHERE collprovider='i')" ).fetchone()[0]: @@ -203,6 +205,52 @@ def test_source_encoding_and_locale_are_copied(provider): assert read_settings(copy.dsn) == discovered +def test_candidate_backend_option_is_rejected_before_postgres_15(workspace, tmp_path): + from typer.testing import CliRunner + + from dbreduce.cli import app + + admin = os.environ["DBREDUCE_TEST_ADMIN"] + with psycopg.connect(admin) as conn: + if conn.info.server_version >= 150000: + pytest.skip("Only servers before PostgreSQL 15 reject the second backend") + before = { + row[0] + for row in conn.execute( + "SELECT datname FROM pg_database WHERE datname LIKE 'dbreduce%'" + ) + } + with pytest.raises(ValueError, match="PostgreSQL 15 or later"): + CloneStore(admin, read_settings(workspace.dsn), Performance()) + result = CliRunner().invoke( + app, + [ + "reduce", + "--database", + workspace.dsn, + "--oracle", + "true", + "--candidate-backend", + "clone", + "--output", + str(tmp_path / "min.sql"), + "--report", + str(tmp_path / "report.json"), + ], + ) + assert result.exit_code == 1 + assert "PostgreSQL 15 or later" in result.output + assert not (tmp_path / "min.sql").exists() + with psycopg.connect(admin) as conn: # the run dropped everything it created + after = { + row[0] + for row in conn.execute( + "SELECT datname FROM pg_database WHERE datname LIKE 'dbreduce%'" + ) + } + assert after == before + + def test_archive_reduction_uses_archived_locale(tmp_path): import json @@ -452,7 +500,6 @@ def test_disconnect_while_create_is_waiting_reports_unverified_database(): cleanup_connecting = threading.Event() thread = None name_was_absent = False - observed_identity = None class CreatorConnection: def __enter__(self): @@ -518,11 +565,6 @@ def run_workspace(): assert errors and isinstance(errors[0], RuntimeError) assert workspace.name in str(errors[0]) assert not workspace.created - with real_connect(admin) as observer: - observed_identity = observer.execute( - "SELECT oid, datdba FROM pg_database WHERE datname = %s", - (workspace.name,), - ).fetchone() finally: try: blocker.execute("ROLLBACK") @@ -549,11 +591,14 @@ def run_workspace(): while cleanup.execute(active_create, (creator_pid, create_query)).fetchone()[0]: assert time.monotonic() < deadline, "CREATE backend did not terminate" time.sleep(0.05) + # The name is random and was absent when the test began; the late CREATE + # may finish after the identity was read, so drop it unless someone marked it. current = cleanup.execute( - "SELECT oid, datdba FROM pg_database WHERE datname = %s", + "SELECT shobj_description(oid, 'pg_database') FROM pg_database " + "WHERE datname = %s", (workspace.name,), ).fetchone() - if observed_identity is not None and current == observed_identity: + if current is not None and current[0] is None: cleanup.execute( sql.SQL("DROP DATABASE {}").format(sql.Identifier(workspace.name)) ) diff --git a/tests/test_workspace.py b/tests/test_workspace.py index 2d8b0a6..4c1b9f8 100644 --- a/tests/test_workspace.py +++ b/tests/test_workspace.py @@ -1,4 +1,5 @@ import os +import signal import subprocess from unittest.mock import MagicMock, patch @@ -93,6 +94,32 @@ def execute(query, *_args): assert statement.startswith('DROP DATABASE "dbreduce_') +def test_leftover_report_does_not_reuse_the_result_for_an_earlier_database(): + connection = MagicMock() + connection.__enter__.return_value = connection + workspace = Workspace("postgresql:///original") + + def execute(query, *_args): + cursor = MagicMock() + statement = query.as_string() if hasattr(query, "as_string") else query + if statement.startswith("SELECT oid, datdba FROM pg_database"): + cursor.fetchone.return_value = (123, 456) + elif statement.startswith("SELECT oid, datdba, shobj_description"): + cursor.fetchone.return_value = (123, 456, workspace.marker) + return cursor + + connection.execute.side_effect = execute + with patch("dbreduce.postgres.database.psycopg.connect", return_value=connection): + workspace.create() + workspace.create() # drops the first database, then creates the second + (leftover,) = workspace.leftovers() + assert leftover.name == workspace.name + assert leftover.state == "removal not attempted" + assert workspace.marker in leftover.detail + workspace.close() + assert workspace.leftovers() == [] + + def test_interrupt_after_create_still_drops_copy(): connection = MagicMock() connection.__enter__.return_value = connection @@ -193,14 +220,14 @@ def test_cleanup_failure_keeps_original_reason_without_leaking_database_error( def test_inline_sslpassword_is_rejected_before_spawning(): - with patch("dbreduce.postgres.dump.subprocess.run") as run: + with patch("dbreduce.postgres.dump.run_owned") as run: with pytest.raises(ValueError, match="sslpassword"): client(["pg_dump"], "dbname=test password=dbsecret sslpassword=keysecret") run.assert_not_called() def test_database_password_not_in_process_arguments(): - with patch("dbreduce.postgres.dump.subprocess.run") as run: + with patch("dbreduce.postgres.dump.run_owned") as run: run.return_value.returncode = 0 client(["pg_dump"], "dbname=test password=syntheticsecret") args = run.call_args.args[0] @@ -210,7 +237,7 @@ def test_database_password_not_in_process_arguments(): def test_database_client_timeout_is_reported(): with patch( - "dbreduce.postgres.dump.subprocess.run", + "dbreduce.postgres.dump.run_owned", side_effect=subprocess.TimeoutExpired("pg_dump", 600), ): with pytest.raises(RuntimeError, match="timed out"): @@ -218,7 +245,7 @@ def test_database_client_timeout_is_reported(): def test_database_client_error_classified_without_leaking_details(): - with patch("dbreduce.postgres.dump.subprocess.run") as run: + with patch("dbreduce.postgres.dump.run_owned") as run: run.return_value.returncode = 1 run.return_value.stderr = "password authentication failed: syntheticsecret" with pytest.raises(RuntimeError, match="authentication failed") as error: @@ -255,7 +282,7 @@ def test_archive_settings_ignore_non_utf8_schema_text(tmp_path, encoding): + f"ENCODING = '{encoding}' LOCALE_PROVIDER = libc LOCALE = 'C';\n".encode() ) - with patch("dbreduce.postgres.database.subprocess.run") as run: + with patch("dbreduce.postgres.database.run_owned") as run: run.return_value.returncode = 0 run.return_value.stdout = stdout settings = read_archive_settings(archive) @@ -267,7 +294,7 @@ def test_archive_settings_ignore_non_utf8_schema_text(tmp_path, encoding): def test_archive_with_newline_database_name_reports_pg_restore_rejection(tmp_path): archive = tmp_path / "source.dump" archive.write_bytes(b"PGDMP") - with patch("dbreduce.postgres.database.subprocess.run") as run: + with patch("dbreduce.postgres.database.run_owned") as run: run.return_value.returncode = 1 run.return_value.stderr = ( b"pg_restore: error: database name contains a newline or carriage return: " @@ -276,3 +303,32 @@ def test_archive_with_newline_database_name_reports_pg_restore_rejection(tmp_pat with pytest.raises(ValueError, match="name containing a newline or carriage return"): read_archive_settings(archive) assert run.call_args.kwargs["env"]["LC_ALL"] == "C" + + +def test_database_already_absent_is_counted_apart_from_dropped(): + connection = MagicMock() + connection.__enter__.return_value = connection + workspace = Workspace("postgresql:///original") + + def execute(query, *_args): + cursor = MagicMock() + statement = query.as_string() if hasattr(query, "as_string") else query + if statement.startswith("SELECT oid, datdba FROM pg_database"): + cursor.fetchone.return_value = (123, 456) + elif statement.startswith("SELECT oid, datdba, shobj_description"): + cursor.fetchone.return_value = None # someone else already removed it + return cursor + + connection.execute.side_effect = execute + with patch("dbreduce.postgres.database.psycopg.connect", return_value=connection): + workspace.create() + workspace.close() + assert not workspace.created and workspace.leftovers() == [] + assert (workspace.databases_dropped, workspace.databases_absent) == (0, 1) + assert workspace.cleanup_failures == 0 + + +def test_kill_signal_exists_on_every_platform(): + from dbreduce.processes import KILL + + assert KILL in (getattr(signal, "SIGKILL", None), signal.SIGTERM) diff --git a/uv.lock b/uv.lock index 44fa5a6..2456a30 100644 --- a/uv.lock +++ b/uv.lock @@ -90,7 +90,7 @@ wheels = [ [[package]] name = "dbreduce" -version = "0.5.0" +version = "0.5.1" source = { editable = "." } dependencies = [ { name = "psycopg", extra = ["binary"] },