diff --git a/CHANGELOG.md b/CHANGELOG.md index f6c72fc..a44d2a0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,33 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +**Upgrading:** No database migration is required. + +### Added + +- Log `worker_connection_hint` once at INFO during `Worker.run()` startup + on PostgreSQL without pooling and on MySQL when `CONN_MAX_AGE = 0`. + The `django_ox` record includes `worker_id`, `database` and `vendor`. + The hint reads settings without opening a connection and does not + prevent startup. + + To retain execution connections, set `CONN_MAX_AGE` above zero and + `CONN_HEALTH_CHECKS = True` in worker-specific settings. + Tasks must restore session state they change. + PostgreSQL's Django pool is another option and requires + `CONN_MAX_AGE = 0`. + See [connection guidance](https://oxpull.com/django-ox/production/#retaining-worker-connections) + for session handling and disconnect recovery. + +### Compatibility + +- Default connection lifecycle, transaction, delivery and schema semantics + are unchanged. Persistence and pooling remain opt-in. + Workers with and without persistence shared a queue in tests using + the same django-ox release. + ## [1.8.0] - 2026-10-04 **Upgrading to 1.8.0:** No database migration is required. diff --git a/docs/llms-full.txt b/docs/llms-full.txt index 5e03584..be90dfa 100644 --- a/docs/llms-full.txt +++ b/docs/llms-full.txt @@ -4826,6 +4826,33 @@ workers. Unpooled renewal and watchdog connects do not gain these local deadlines. PgBouncer and third-party pools have not been tested. +### Retaining worker connections + +With `CONN_MAX_AGE = 0` and no pool, Django closes initialized execution connections before and after each attempt. The next attempt opens another connection; the poll loop keeps its claim connection. Task ORM queries and outcome writing use the same execution connection on the worker's database alias. `worker_connection_hint` reports this setting once at INFO during `Worker.run()` startup on PostgreSQL and MySQL. + +To retain connections, configure the worker's alias in a dedicated worker settings module. Launch workers with that module to leave web processes unchanged. For PostgreSQL or MySQL without a pool, set a positive connection age and enable health checks: + +```python +DATABASES["default"]["CONN_MAX_AGE"] = 600 +DATABASES["default"]["CONN_HEALTH_CHECKS"] = True +``` + +For PostgreSQL, Django's pool requires psycopg 3, `psycopg_pool` and `CONN_MAX_AGE = 0`; use the pool example above instead of a positive age. MySQL has no Django pool; setting `OPTIONS["pool"]` does not enable one. + +Allow pool capacity for task threads and polling, plus the separate renewal, watchdog and stop-time recovery connections described above. Each worker process has its own pool. Retained connections carry session state between tasks; follow the responsibilities below. + +Persistence and pooling remain opt-in. The default connection lifecycle, transactions, delivery and schema are unchanged. + +### Session state and connection failures + +With persistence or pooling, tasks must account for session settings, temporary objects and session locks carried over from earlier tasks. Finish transaction scopes and restore any session state you change. Django's age, usability and autocommit checks are not a comprehensive session reset. With `CONN_MAX_AGE = 0` and no pool, a later task does not inherit that execution session. + +Each execution thread keeps its own connection context, separate from other task threads, claiming and renewal. Sharing an execution connection does not make task effects and outcome writing one atomic transaction. Delivery remains at least once with lease fencing; tasks must tolerate retry. + +Measured default runs opened one connection per task attempt plus one for polling; ORM queries added no connections. Persistence or PostgreSQL pooling opened two connections in total: one for execution and one for polling, regardless of task count. These counts are exact on PostgreSQL and SQLite; MySQL counting tests allow one extra connection per thread because Django's MySQL backend reads the server version through a throwaway connection the first time a thread uses it. After a server-ended idle session, querying tasks used one extra attempt, then succeeded on a new session. Query-free tasks reconnected at outcome writing without spending an attempt; every task completed. Workers with and without persistence shared a queue: all 40 tasks succeeded, each running once with one claim and one epoch. Both workers ran tasks. These measurements cover same-release batch workers, PostgreSQL, MySQL and SQLite connection counts, PostgreSQL pooling, and PostgreSQL and MySQL idle-session recovery. + +Set `CONN_HEALTH_CHECKS = True`; health checks reduce stale-connection failures but cannot prevent later disconnects. + ### psycopg_pool versions and recovery Use psycopg_pool 3.3.3 or newer with Django's PostgreSQL pool. Version @@ -6232,6 +6259,7 @@ The message text is not part of the contract. The keys are. | `worker_started` | INFO | The run loop starts. | | `connection_pool_too_small` | WARNING | Once per `Worker.run()`, after `worker_started` and before threads start: the worker alias's effective PostgreSQL pool maximum is below `concurrency + 1`. Each `--processes` child checks separately. The warning does not resize the pool, refuse startup, measure available server slots, or reserve fallback capacity. | | `connection_pool_cannot_drain` | WARNING | Pooled PostgreSQL only. The installed psycopg_pool has no `ConnectionPool.drain()` method, which was added in 3.3.0. Emitted once per `Worker.run()`, after `worker_started` and the `connection_pool_too_small` check, before threads start. Each `--processes` child checks separately. Also emitted at the first sweep of such a pool in a process, for each database alias, unless startup already emitted it in that process. Concurrent first sweeps may emit duplicate warnings. The sweep record is the only notice available to callers of `run_once()` or `run_tasks()`. Idle connections are still tested as in django-ox 1.7.0, preserving restart recovery but also the wait on a connection that stays open without answering. A stop is not read until that test returns. Upgrade psycopg_pool to 3.3.3 or newer. This warning does not refuse startup or change the pool. No traceback. Keys: `event` and `database` always; `worker_id` only on the startup record. | +| `worker_connection_hint` | INFO | Once at `Worker.run()` startup on PostgreSQL and MySQL when `CONN_MAX_AGE = 0` and no PostgreSQL pool is configured. MySQL ignores pool options. Each attempt opens one execution connection for task queries and outcome writing. Carries `worker_id`, `database` and `vendor`. The hint opens no connection and never stops the worker. See [connection guidance](production.md#retaining-worker-connections). | | `task_claimed` | DEBUG | A task was claimed from the queue. | | `task_started` | DEBUG | Execution of an attempt begins. | | `task_succeeded` | INFO | The task reached SUCCESSFUL. | @@ -6358,7 +6386,8 @@ A failed connect while recording a stuck attempt logs | `reason` | `schedule_row_skipped` | Why the row could not be used, during either a full read or a locked dispatch read. For a value that could not be converted, the reason names the column. | | `queues` | `worker_started` | The worker's queues. | | `concurrency` | `worker_started`, `connection_pool_too_small` | The worker's task-thread concurrency, set by `--concurrency`. | -| `database` | `connection_pool_too_small`, `schedule_dispatch_error`, `schedule_dispatch_failed`, `schedule_dispatch_recovered` | The worker's database alias: `--database`, or the alias used to write `OxTask`. | +| `database` | `connection_pool_too_small`, `worker_connection_hint`, `schedule_dispatch_error`, `schedule_dispatch_failed`, `schedule_dispatch_recovered` | The worker's database alias: `--database`, or the alias used to write `OxTask`. | +| `vendor` | `worker_connection_hint` | Django's name for the database engine, such as `postgresql` or `mysql`. | | `max_size` | `connection_pool_too_small` | The effective pool maximum. `pool=True` means 4. For a non-empty mapping, use `max_size`; if absent or `None`, use `min_size`, defaulting to 4. The sizing check skips values that are not an `int` of at least 1. It excludes booleans and floats such as `10.0`. | | `recommended_max_size` | `connection_pool_too_small` | `concurrency + 1`: one pooled connection per task thread and one for the poll loop. This does not reserve fallback capacity or validate the server budget. | | `unpooled_connections` | `connection_pool_too_small` | Worst-case additional private-connection budget per worker process in normal operation, not a count of open connections. Always 2: one for lease renewal and one possible timeout-watchdog connection. This does not count the additional private connection used for a stop-time recovery look. A worker whose tasks never use timeouts needs only the first. An absent timeout in `OPTIONS` alone does not establish that, because a task can declare its own. | @@ -7687,6 +7716,33 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +**Upgrading:** No database migration is required. + +### Added + +- Log `worker_connection_hint` once at INFO during `Worker.run()` startup + on PostgreSQL without pooling and on MySQL when `CONN_MAX_AGE = 0`. + The `django_ox` record includes `worker_id`, `database` and `vendor`. + The hint reads settings without opening a connection and does not + prevent startup. + + To retain execution connections, set `CONN_MAX_AGE` above zero and + `CONN_HEALTH_CHECKS = True` in worker-specific settings. + Tasks must restore session state they change. + PostgreSQL's Django pool is another option and requires + `CONN_MAX_AGE = 0`. + See [connection guidance](https://oxpull.com/django-ox/production/#retaining-worker-connections) + for session handling and disconnect recovery. + +### Compatibility + +- Default connection lifecycle, transaction, delivery and schema semantics + are unchanged. Persistence and pooling remain opt-in. + Workers with and without persistence shared a queue in tests using + the same django-ox release. + ## [1.8.0] - 2026-10-04 **Upgrading to 1.8.0:** No database migration is required. diff --git a/docs/monitoring.md b/docs/monitoring.md index 3be4291..9771f1e 100644 --- a/docs/monitoring.md +++ b/docs/monitoring.md @@ -669,6 +669,7 @@ The message text is not part of the contract. The keys are. | `worker_started` | INFO | The run loop starts. | | `connection_pool_too_small` | WARNING | Once per `Worker.run()`, after `worker_started` and before threads start: the worker alias's effective PostgreSQL pool maximum is below `concurrency + 1`. Each `--processes` child checks separately. The warning does not resize the pool, refuse startup, measure available server slots, or reserve fallback capacity. | | `connection_pool_cannot_drain` | WARNING | Pooled PostgreSQL only. The installed psycopg_pool has no `ConnectionPool.drain()` method, which was added in 3.3.0. Emitted once per `Worker.run()`, after `worker_started` and the `connection_pool_too_small` check, before threads start. Each `--processes` child checks separately. Also emitted at the first sweep of such a pool in a process, for each database alias, unless startup already emitted it in that process. Concurrent first sweeps may emit duplicate warnings. The sweep record is the only notice available to callers of `run_once()` or `run_tasks()`. Idle connections are still tested as in django-ox 1.7.0, preserving restart recovery but also the wait on a connection that stays open without answering. A stop is not read until that test returns. Upgrade psycopg_pool to 3.3.3 or newer. This warning does not refuse startup or change the pool. No traceback. Keys: `event` and `database` always; `worker_id` only on the startup record. | +| `worker_connection_hint` | INFO | Once at `Worker.run()` startup on PostgreSQL and MySQL when `CONN_MAX_AGE = 0` and no PostgreSQL pool is configured. MySQL ignores pool options. Each attempt opens one execution connection for task queries and outcome writing. Carries `worker_id`, `database` and `vendor`. The hint opens no connection and never stops the worker. See [connection guidance](production.md#retaining-worker-connections). | | `task_claimed` | DEBUG | A task was claimed from the queue. | | `task_started` | DEBUG | Execution of an attempt begins. | | `task_succeeded` | INFO | The task reached SUCCESSFUL. | @@ -795,7 +796,8 @@ A failed connect while recording a stuck attempt logs | `reason` | `schedule_row_skipped` | Why the row could not be used, during either a full read or a locked dispatch read. For a value that could not be converted, the reason names the column. | | `queues` | `worker_started` | The worker's queues. | | `concurrency` | `worker_started`, `connection_pool_too_small` | The worker's task-thread concurrency, set by `--concurrency`. | -| `database` | `connection_pool_too_small`, `schedule_dispatch_error`, `schedule_dispatch_failed`, `schedule_dispatch_recovered` | The worker's database alias: `--database`, or the alias used to write `OxTask`. | +| `database` | `connection_pool_too_small`, `worker_connection_hint`, `schedule_dispatch_error`, `schedule_dispatch_failed`, `schedule_dispatch_recovered` | The worker's database alias: `--database`, or the alias used to write `OxTask`. | +| `vendor` | `worker_connection_hint` | Django's name for the database engine, such as `postgresql` or `mysql`. | | `max_size` | `connection_pool_too_small` | The effective pool maximum. `pool=True` means 4. For a non-empty mapping, use `max_size`; if absent or `None`, use `min_size`, defaulting to 4. The sizing check skips values that are not an `int` of at least 1. It excludes booleans and floats such as `10.0`. | | `recommended_max_size` | `connection_pool_too_small` | `concurrency + 1`: one pooled connection per task thread and one for the poll loop. This does not reserve fallback capacity or validate the server budget. | | `unpooled_connections` | `connection_pool_too_small` | Worst-case additional private-connection budget per worker process in normal operation, not a count of open connections. Always 2: one for lease renewal and one possible timeout-watchdog connection. This does not count the additional private connection used for a stop-time recovery look. A worker whose tasks never use timeouts needs only the first. An absent timeout in `OPTIONS` alone does not establish that, because a task can declare its own. | diff --git a/docs/production.md b/docs/production.md index f1f1852..d75a63d 100644 --- a/docs/production.md +++ b/docs/production.md @@ -868,6 +868,33 @@ workers. Unpooled renewal and watchdog connects do not gain these local deadlines. PgBouncer and third-party pools have not been tested. +### Retaining worker connections + +With `CONN_MAX_AGE = 0` and no pool, Django closes initialized execution connections before and after each attempt. The next attempt opens another connection; the poll loop keeps its claim connection. Task ORM queries and outcome writing use the same execution connection on the worker's database alias. `worker_connection_hint` reports this setting once at INFO during `Worker.run()` startup on PostgreSQL and MySQL. + +To retain connections, configure the worker's alias in a dedicated worker settings module. Launch workers with that module to leave web processes unchanged. For PostgreSQL or MySQL without a pool, set a positive connection age and enable health checks: + +```python +DATABASES["default"]["CONN_MAX_AGE"] = 600 +DATABASES["default"]["CONN_HEALTH_CHECKS"] = True +``` + +For PostgreSQL, Django's pool requires psycopg 3, `psycopg_pool` and `CONN_MAX_AGE = 0`; use the pool example above instead of a positive age. MySQL has no Django pool; setting `OPTIONS["pool"]` does not enable one. + +Allow pool capacity for task threads and polling, plus the separate renewal, watchdog and stop-time recovery connections described above. Each worker process has its own pool. Retained connections carry session state between tasks; follow the responsibilities below. + +Persistence and pooling remain opt-in. The default connection lifecycle, transactions, delivery and schema are unchanged. + +### Session state and connection failures + +With persistence or pooling, tasks must account for session settings, temporary objects and session locks carried over from earlier tasks. Finish transaction scopes and restore any session state you change. Django's age, usability and autocommit checks are not a comprehensive session reset. With `CONN_MAX_AGE = 0` and no pool, a later task does not inherit that execution session. + +Each execution thread keeps its own connection context, separate from other task threads, claiming and renewal. Sharing an execution connection does not make task effects and outcome writing one atomic transaction. Delivery remains at least once with lease fencing; tasks must tolerate retry. + +Measured default runs opened one connection per task attempt plus one for polling; ORM queries added no connections. Persistence or PostgreSQL pooling opened two connections in total: one for execution and one for polling, regardless of task count. These counts are exact on PostgreSQL and SQLite; MySQL counting tests allow one extra connection per thread because Django's MySQL backend reads the server version through a throwaway connection the first time a thread uses it. After a server-ended idle session, querying tasks used one extra attempt, then succeeded on a new session. Query-free tasks reconnected at outcome writing without spending an attempt; every task completed. Workers with and without persistence shared a queue: all 40 tasks succeeded, each running once with one claim and one epoch. Both workers ran tasks. These measurements cover same-release batch workers, PostgreSQL, MySQL and SQLite connection counts, PostgreSQL pooling, and PostgreSQL and MySQL idle-session recovery. + +Set `CONN_HEALTH_CHECKS = True`; health checks reduce stale-connection failures but cannot prevent later disconnects. + ### psycopg_pool versions and recovery Use psycopg_pool 3.3.3 or newer with Django's PostgreSQL pool. Version diff --git a/src/django_ox/worker.py b/src/django_ox/worker.py index 1e8d4f5..164b129 100644 --- a/src/django_ox/worker.py +++ b/src/django_ox/worker.py @@ -5858,6 +5858,40 @@ def _warn_if_the_connection_pool_is_short(self) -> None: }, ) + def _hint_if_every_task_reconnects(self) -> None: + """ + Report per-attempt execution connections at PostgreSQL and MySQL startup. + + Log worker_connection_hint once at INFO when CONN_MAX_AGE is zero and + PostgreSQL pooling is absent. Read settings without opening a connection + or preventing worker startup. + """ + conn = connections[self._db_alias] + if conn.vendor not in ("postgresql", "mysql"): + return + if conn.settings_dict.get("CONN_MAX_AGE", 0) != 0: + return + if conn.vendor == "postgresql" and _pool_options(self._db_alias) is not None: + return + logger.info( + "Worker %s: database %r opens a new execution connection per " + "attempt with CONN_MAX_AGE = 0 on PostgreSQL or MySQL. " + "To retain connections, set CONN_MAX_AGE above 0 and " + "CONN_HEALTH_CHECKS = True in worker settings. " + "Tasks must restore session state they change; a server-ended " + "idle session can cost a querying task an extra attempt. " + "PostgreSQL's Django pool is another option and requires " + "CONN_MAX_AGE = 0.", + self.worker_id, + self._db_alias, + extra={ + "event": "worker_connection_hint", + "worker_id": self.worker_id, + "database": self._db_alias, + "vendor": conn.vendor, + }, + ) + def _warn_if_the_connection_pool_cannot_drain(self) -> None: """ Say so at startup when Django's PostgreSQL connection pool for this @@ -6061,6 +6095,7 @@ def run(self) -> None: ) self._warn_if_the_connection_pool_is_short() self._warn_if_the_connection_pool_cannot_drain() + self._hint_if_every_task_reconnects() in_flight: set[Future[None]] = set() last_reap = 0.0 last_dispatch = 0.0 diff --git a/tests/connection_count_tasks.py b/tests/connection_count_tasks.py new file mode 100644 index 0000000..6818b7c --- /dev/null +++ b/tests/connection_count_tasks.py @@ -0,0 +1,31 @@ +"""Tasks for test_connection_count. Each notes what ran, where, to a file.""" + +import time + +from django.contrib.auth.models import User + +from django_ox.compat import task + +from .dead_connection_tasks import note, session_id + + +@task +def tick(notes, seconds=0.0, index=None): + note(notes, "ran", index=index) + if seconds: + time.sleep(seconds) + return "ran" + + +@task +def count_users(notes): + note(notes, "ran", users=User.objects.count()) + return "ran" + + +@task +def report_session(notes, seconds=0.0): + note(notes, "ran", session=session_id()) + if seconds: + time.sleep(seconds) + return "ran" diff --git a/tests/connection_counter.py b/tests/connection_counter.py new file mode 100644 index 0000000..feea0ed --- /dev/null +++ b/tests/connection_counter.py @@ -0,0 +1,83 @@ +""" +Count the connections a process opens to its database, for +test_connection_count. + +Django's connect hooks fire for the connections it makes and say nothing about +the ones a pool or a driver makes on its own, so this wraps the driver call +itself. Every call appends one line, the process id and the name of the thread +that made it, to a file; a line is written with one os.write so that threads +and processes do not interleave. It is installed by the settings module of a +worker the test starts, before Django opens anything, and is never imported by +the package. +""" + +import os +import threading +from collections import Counter +from pathlib import Path + + +def install(path: str, vendor: str) -> None: + fd = os.open(path, os.O_WRONLY | os.O_APPEND | os.O_CREAT) + + def note() -> None: + line = f"{os.getpid()} {threading.current_thread().name}\n" + os.write(fd, line.encode()) + + if vendor == "postgresql": + import psycopg + + connect = psycopg.Connection.connect.__func__ # type: ignore[attr-defined] + plain = psycopg.connect + + def counted_class_connect(cls, *args, **kwargs): # type: ignore[no-untyped-def] + note() + return connect(cls, *args, **kwargs) + + def counted_connect(*args, **kwargs): # type: ignore[no-untyped-def] + note() + return plain(*args, **kwargs) + + psycopg.Connection.connect = classmethod(counted_class_connect) # type: ignore[method-assign,assignment] + psycopg.connect = counted_connect # type: ignore[assignment] + elif vendor == "mysql": + import pymysql + + pymysql.install_as_MySQLdb() + original = pymysql.connect + + def counted(*args, **kwargs): # type: ignore[no-untyped-def] + note() + return original(*args, **kwargs) + + pymysql.connect = counted # type: ignore[assignment] + pymysql.Connect = counted # type: ignore[assignment] + else: + # Django calls sqlite3.dbapi2.connect, not sqlite3.connect: wrapping + # the second counts nothing. + from sqlite3 import dbapi2 + + original_sqlite = dbapi2.connect + + def counted_sqlite(*args, **kwargs): # type: ignore[no-untyped-def] + note() + return original_sqlite(*args, **kwargs) + + dbapi2.connect = counted_sqlite # type: ignore[assignment] + + +def read(path: Path | str) -> list[tuple[int, str]]: + """The (process id, thread name) of every connect call, in order.""" + found = Path(path) + if not found.exists(): + return [] + rows = [] + for line in found.read_text().splitlines(): + pid, _, thread = line.partition(" ") + rows.append((int(pid), thread)) + return rows + + +def by_thread(path: Path | str) -> Counter[str]: + """Connect calls per thread name.""" + return Counter(thread for _, thread in read(path)) diff --git a/tests/test_connection_count.py b/tests/test_connection_count.py new file mode 100644 index 0000000..4018806 --- /dev/null +++ b/tests/test_connection_count.py @@ -0,0 +1,566 @@ +""" +How many database connections `ox_worker` opens, and the hint it logs when +every task attempt opens one. + +At the default CONN_MAX_AGE of 0 Django closes a thread's connections after +each use, and a task attempt is one use: the execution thread opens a new +connection for every attempt, and the task's queries and its outcome write +ride that one. The poll loop's connection is kept for the life of the process. +With CONN_MAX_AGE set, or with Django's PostgreSQL pool, the execution thread +keeps its connection. Nothing here changes either behaviour; these tests +state them and fail if a change moves them. + +Every test runs the real `ox_worker --batch` command in a process of its own +against tasks queued beforehand. The driver's connect call is wrapped in that +process (tests/connection_counter.py), so the count is the connections the +worker opened, by thread, and not the ones the test process opens to watch the +rows. PostgreSQL's own session counter is read as a second witness. +""" + +import contextlib +import copy +import importlib.util +import json +import logging +import os +import subprocess +import sys +import time +from collections import Counter +from pathlib import Path + +import pytest +from django.conf import settings +from django.db import connection, connections + +from django_ox.models import OxTask +from django_ox.worker import Worker + +from . import connection_counter +from .connection_count_tasks import count_users, report_session, tick +from .dead_connection_tasks import end_session, read_notes + +REPO = Path(__file__).resolve().parent.parent + +pytestmark = [ + pytest.mark.django_db(transaction=True), + pytest.mark.skipif(os.name != "posix", reason="runs the worker as a process"), +] + +on_postgresql = pytest.mark.skipif( + connection.vendor != "postgresql", reason="counts PostgreSQL sessions" +) +ends_a_connection = pytest.mark.skipif( + connection.vendor not in ("postgresql", "mysql"), + reason="has the server end a connection, with pg_terminate_backend or KILL", +) +with_psycopg_pool = pytest.mark.skipif( + connection.vendor != "postgresql" + or importlib.util.find_spec("psycopg_pool") is None, + reason="Django's pool is PostgreSQL only and needs psycopg_pool", +) + +#: Small enough for CI, large enough that one connection per attempt cannot be +#: mistaken for a fixed cost. +TASKS = 30 + +POOL = {"min_size": 1, "max_size": 6} +LIMIT = 90.0 +HINT = "worker_connection_hint" +HINT_TEXT = ( + "Worker %s: database %r opens a new execution connection per " + "attempt with CONN_MAX_AGE = 0 on PostgreSQL or MySQL. " + "To retain connections, set CONN_MAX_AGE above 0 and " + "CONN_HEALTH_CHECKS = True in worker settings. " + "Tasks must restore session state they change; a server-ended " + "idle session can cost a querying task an extra attempt. " + "PostgreSQL's Django pool is another option and requires " + "CONN_MAX_AGE = 0." +) + + +def project(tmp_path: Path, mode: str, name: str = "w") -> tuple[Path, Path]: + """ + A project whose settings are this run's with the default database in + `mode`: "default" (CONN_MAX_AGE 0, no pool), "persistent" (CONN_MAX_AGE + 600) or "pool" (Django's PostgreSQL pool). Returns it with the file its + worker records every connect call in. + """ + root = tmp_path / name + (root / "countproj").mkdir(parents=True) + (root / "manage.py").write_text( + "import os, sys\n" + "os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'countproj.settings')\n" + "from django.core.management import execute_from_command_line\n" + "execute_from_command_line(sys.argv)\n" + ) + (root / "countproj" / "__init__.py").write_text("") + counts = tmp_path / f"{name}.connects" + lines = [ + "import sys", + f"sys.path[:0] = [{str(REPO / 'src')!r}, {str(REPO)!r}]", + f"from {settings.SETTINGS_MODULE} import * # noqa: F403", + "from tests.connection_counter import install", + f"install({str(counts)!r}, {connection.vendor!r})", + "_db = DATABASES['default']", + "_opts = _db.get('OPTIONS', {})", + "_db['OPTIONS'] = {k: v for k, v in _opts.items() if k != 'pool'}", + f"_db['CONN_MAX_AGE'] = {600 if mode == 'persistent' else 0}", + ] + if mode == "pool": + lines.append(f"_db['OPTIONS']['pool'] = {POOL!r}") + (root / "countproj" / "settings.py").write_text("\n".join(lines) + "\n") + return root, counts + + +def start( + root: Path, + tmp_path: Path, + *flags: str, + name: str = "w", + options: dict | None = None, +): + env = dict(os.environ) + env["DJANGO_SETTINGS_MODULE"] = "countproj.settings" + env["OX_TEST_DB_NAME"] = str(connection.settings_dict["NAME"]) + env["OX_TEST_LOG_LEVEL"] = "INFO" + env["OX_TEST_LOG_FORMAT"] = "%(event)s %(message)s" + if options is not None: + env["OX_TEST_TASKS_OPTIONS"] = json.dumps(options) + log = tmp_path / f"{name}.log" + with log.open("wb") as out: + proc = subprocess.Popen( # noqa: S603 + [sys.executable, "manage.py", "ox_worker", "--interval", "0.05", *flags], + cwd=root, + env=env, + stdout=out, + stderr=out, + ) + return proc, log + + +def finish(proc: subprocess.Popen, log: Path) -> str: + try: + code = proc.wait(timeout=LIMIT) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait(timeout=10) + pytest.fail(f"the worker did not exit by itself:\n{log.read_text()}") + output = log.read_text() + assert code == 0, output + return output + + +def run_batch(tmp_path: Path, mode: str, *flags: str) -> tuple[Path, str]: + """Run `ox_worker --batch` in `mode`; the connect log and the output.""" + root, counts = project(tmp_path, mode) + proc, log = start(root, tmp_path, "--batch", "--concurrency", "1", *flags) + return counts, finish(proc, log) + + +def hints(output: str) -> list[str]: + return [line for line in output.splitlines() if line.startswith(f"{HINT} ")] + + +def sessions_opened() -> int | None: + """PostgreSQL's count of sessions it has started on the test database.""" + if connection.vendor != "postgresql": + return None + with connection.cursor() as cursor: + cursor.execute("SELECT pg_stat_force_next_flush()") + cursor.execute( + "SELECT sessions FROM pg_stat_database WHERE datname = current_database()" + ) + return int(cursor.fetchone()[0]) + + +#: Django's MySQL backend reads the server's version once per thread's +#: wrapper, and when that is the thread's first use it does so on a connection +#: of its own, opened and closed before the ordinary one. It is a cost of the +#: first attempt on a thread, not of every attempt, so a count on MySQL may +#: exceed the expected one by up to the number of threads that ran tasks. +EXTRA_ON_FIRST_USE = 1 if connection.vendor == "mysql" else 0 + + +def assert_opened(found, expected, output): + """`found` connections where `expected` are, give or take the one above.""" + assert expected <= found <= expected + EXTRA_ON_FIRST_USE, ( + f"{found} connections, expected {expected}\n{output}" + ) + + +def all_succeeded_once(): + rows = list(OxTask.objects.all()) + assert Counter(row.status for row in rows) == {OxTask.Status.SUCCESSFUL: len(rows)} + assert {row.attempts for row in rows} == {1} + + +# -- the counts ----------------------------------------------------------------- + + +def execution_threads(counts: Path) -> Counter[str]: + """Connects by thread, with every execution thread under one name.""" + return Counter( + "execution" if name.startswith("ox_") else name + for _, name in connection_counter.read(counts) + ) + + +class TestConnectionsPerAttempt: + def test_zero_tasks_cost_one_connection(self, tmp_path): + """The fixed cost of starting a worker, which every count below adds to.""" + counts, output = run_batch(tmp_path, "default") + assert execution_threads(counts) == {"MainThread": 1}, output + + @pytest.mark.skipif( + connection.vendor == "sqlite", + reason="SQLite opens a file, not a server connection, and logs no hint", + ) + def test_the_default_opens_one_connection_per_attempt(self, tmp_path): + before = sessions_opened() + for i in range(TASKS): + tick.enqueue(str(tmp_path / "n.jsonl"), index=i) + counts, output = run_batch(tmp_path, "default") + all_succeeded_once() + found = execution_threads(counts) + # The poll loop keeps the connection it started with; each attempt's + # thread opens its own and closes it when the attempt ends. + assert found["MainThread"] == 1, output + assert_opened(found["execution"], TASKS, output) + assert_opened(sum(found.values()), TASKS + 1, output) + if before is not None: + assert sessions_opened() - before == TASKS + 1 + + @pytest.mark.skipif( + connection.vendor != "sqlite", reason="the SQLite arm of the same count" + ) + def test_sqlite_default_also_opens_one_per_attempt(self, tmp_path): + for i in range(TASKS): + tick.enqueue(str(tmp_path / "n.jsonl"), index=i) + counts, output = run_batch(tmp_path, "default") + all_succeeded_once() + assert execution_threads(counts) == { + "MainThread": 1, + "execution": TASKS, + }, output + assert not hints(output) + + @pytest.mark.parametrize("mode", ["persistent", "pool"]) + def test_a_kept_connection_is_opened_once(self, tmp_path, mode): + if mode == "pool" and ( + connection.vendor != "postgresql" + or importlib.util.find_spec("psycopg_pool") is None + ): + pytest.skip("Django's pool is PostgreSQL only and needs psycopg_pool") + before = sessions_opened() + for i in range(TASKS): + tick.enqueue(str(tmp_path / "n.jsonl"), index=i) + counts, output = run_batch(tmp_path, mode) + all_succeeded_once() + found = execution_threads(counts) + if mode == "persistent": + assert found["MainThread"] == 1, output + assert_opened(found["execution"], 1, output) + else: + # The pool opens its connections on threads of its own. + assert found["execution"] == 0, output + assert_opened(sum(found.values()), 2, output) + assert not hints(output) + if before is not None: + assert sessions_opened() - before == 2 + + def test_task_queries_add_no_connections(self, tmp_path): + """The ORM work and the outcome write share the attempt's connection.""" + for _ in range(TASKS): + count_users.enqueue(str(tmp_path / "n.jsonl")) + counts, output = run_batch(tmp_path, "default") + all_succeeded_once() + found = execution_threads(counts) + assert_opened(sum(found.values()), TASKS + 1, output) + + def test_closing_and_reopening_is_counted_each_time(self, tmp_path): + """ + The negative control: a process that closes its connection and opens + another five times is seen to, so a count that stays low is not a + counter that cannot see. + """ + root, counts = project(tmp_path, "persistent") + before = sessions_opened() + env = dict(os.environ) + env["DJANGO_SETTINGS_MODULE"] = "countproj.settings" + env["OX_TEST_DB_NAME"] = str(connection.settings_dict["NAME"]) + script = ( + "import django; django.setup()\n" + "from django.db import connection\n" + "for _ in range(5):\n" + " connection.close(); connection.ensure_connection()\n" + ) + subprocess.run( # noqa: S603 + [sys.executable, "-c", script], cwd=root, env=env, check=True, timeout=60 + ) + assert execution_threads(counts) == {"MainThread": 5} + if before is not None: + assert sessions_opened() - before == 5 + + +# -- the hint ------------------------------------------------------------------- + + +@pytest.mark.skipif( + connection.vendor == "sqlite", reason="SQLite opens a file and gets no hint" +) +class TestTheHintOnTheRealCommand: + def test_the_default_logs_it_once(self, tmp_path): + tick.enqueue(str(tmp_path / "n.jsonl")) + _, output = run_batch(tmp_path, "default") + assert len(hints(output)) == 1, output + + @pytest.mark.parametrize("mode", ["persistent", "pool"]) + def test_a_kept_connection_logs_none(self, tmp_path, mode): + if mode == "pool" and ( + connection.vendor != "postgresql" + or importlib.util.find_spec("psycopg_pool") is None + ): + pytest.skip("Django's pool is PostgreSQL only and needs psycopg_pool") + tick.enqueue(str(tmp_path / "n.jsonl")) + _, output = run_batch(tmp_path, mode) + assert not hints(output), output + + def test_it_does_not_stop_a_worker_with_nothing_to_do(self, tmp_path): + _, output = run_batch(tmp_path, "default") + assert len(hints(output)) == 1, output + assert "worker_stopped" in output + + +ALIAS = "hinted" + + +@pytest.fixture +def add_alias(): + def add(**overrides): + base = copy.deepcopy(connections.settings["default"]) + connections.settings[ALIAS] = {**base, **overrides} + + yield add + # The wrapper is built once per thread and kept, so a later test would + # get the first one's engine. Without the engine's driver installed no + # wrapper was built, and the delete raises AttributeError. + with contextlib.suppress(KeyError, AttributeError): + del connections[ALIAS] + connections.settings.pop(ALIAS, None) + + +POSTGRESQL = "django.db.backends.postgresql" +MYSQL = "django.db.backends.mysql" +SQLITE = "django.db.backends.sqlite3" + + +def needs_driver(vendor): + """Skip unless Django could build a wrapper for `vendor`.""" + if vendor == "postgresql": + pytest.importorskip("psycopg") + return + if importlib.util.find_spec("MySQLdb") is None: + pymysql = pytest.importorskip("pymysql") + pymysql.install_as_MySQLdb() + + +def hint_records(caplog): + return [r for r in caplog.records if getattr(r, "event", None) == HINT] + + +class TestWhenTheHintIsGiven: + """ + The rule, on aliases that are never connected to: it reads settings and the + engine's name. An alias is a PostgreSQL or MySQL one when the driver is + importable. + """ + + def hint(self, caplog, **alias): + worker = Worker(db_alias=ALIAS) + with caplog.at_level(logging.INFO, logger="django_ox"): + worker._hint_if_every_task_reconnects() + return hint_records(caplog) + + @pytest.mark.parametrize( + ("engine", "vendor"), [(POSTGRESQL, "postgresql"), (MYSQL, "mysql")] + ) + def test_closing_after_use_without_a_pool_gets_it( + self, add_alias, caplog, engine, vendor + ): + needs_driver(vendor) + add_alias(ENGINE=engine, CONN_MAX_AGE=0, OPTIONS={}, TEST={}) + (record,) = self.hint(caplog) + assert record.levelno == logging.INFO + assert record.database == ALIAS + assert record.vendor == vendor + assert record.worker_id + assert record.getMessage() == HINT_TEXT % (record.worker_id, ALIAS) + + def test_a_missing_setting_is_the_default_and_gets_it(self, add_alias, caplog): + pytest.importorskip("psycopg") + add_alias(ENGINE=POSTGRESQL, OPTIONS={}, TEST={}) + connections.settings[ALIAS].pop("CONN_MAX_AGE", None) + assert len(self.hint(caplog)) == 1 + + @pytest.mark.parametrize("age", [1, 600, None]) + def test_keeping_connections_open_gets_none(self, add_alias, caplog, age): + pytest.importorskip("psycopg") + add_alias(ENGINE=POSTGRESQL, CONN_MAX_AGE=age, OPTIONS={}, TEST={}) + assert not self.hint(caplog) + + @pytest.mark.parametrize("pool", [True, {"max_size": 3}]) + def test_a_pool_gets_none(self, add_alias, caplog, pool): + pytest.importorskip("psycopg") + add_alias(ENGINE=POSTGRESQL, CONN_MAX_AGE=0, OPTIONS={"pool": pool}, TEST={}) + assert not self.hint(caplog) + + @pytest.mark.parametrize("pool", [None, False, {}]) + def test_an_option_that_is_not_a_pool_does_not_count(self, add_alias, caplog, pool): + pytest.importorskip("psycopg") + add_alias(ENGINE=POSTGRESQL, CONN_MAX_AGE=0, OPTIONS={"pool": pool}, TEST={}) + assert len(self.hint(caplog)) == 1 + + def test_mysql_has_no_pool_to_set(self, add_alias, caplog): + needs_driver("mysql") + add_alias(ENGINE=MYSQL, CONN_MAX_AGE=0, OPTIONS={"pool": True}, TEST={}) + assert len(self.hint(caplog)) == 1 + + def test_sqlite_gets_none(self, add_alias, caplog): + add_alias(ENGINE=SQLITE, CONN_MAX_AGE=0, OPTIONS={}, TEST={}) + assert not self.hint(caplog) + + def test_an_unsupported_vendor_gets_none(self, add_alias, caplog, monkeypatch): + add_alias(ENGINE=SQLITE, CONN_MAX_AGE=0, OPTIONS={}, TEST={}) + monkeypatch.setattr(connections[ALIAS], "vendor", "oracle") + assert not self.hint(caplog) + + def test_it_opens_no_connection(self, add_alias, caplog): + pytest.importorskip("psycopg") + add_alias( + ENGINE=POSTGRESQL, + CONN_MAX_AGE=0, + OPTIONS={}, + HOST="127.0.0.1", + PORT="1", + TEST={}, + ) + assert len(self.hint(caplog)) == 1 + assert connections[ALIAS].connection is None + + +# -- failure and shared queues -------------------------------------------------- + + +def wait_until(predicate, timeout=60.0): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return + time.sleep(0.05) + pytest.fail("the condition never held") + + +@ends_a_connection +@pytest.mark.parametrize("queries", [False, True], ids=["no-query", "query"]) +def test_a_persistent_worker_recovers_when_its_idle_session_is_ended(tmp_path, queries): + """ + The connection a persistent worker keeps between tasks is ended by the + server while it is idle, as a failover or an idle timeout does. Nothing + flagged it, so the next task finds it dead. The worker opens a new one, + and every task still completes once its attempts allow. + """ + notes = str(tmp_path / "notes.jsonl") + root, counts = project(tmp_path, "persistent") + # Claims, not tasks: the retry of the one attempt that fails is a ninth. + claims = "9" if queries else "8" + proc, log = start( + root, + tmp_path, + "--max-tasks", + claims, + "--concurrency", + "1", + options={"BACKOFF_INITIAL": 0.05, "BACKOFF_MAX": 0.05}, + ) + try: + first = [report_session.enqueue(notes) for _ in range(3)] + wait_until( + lambda: all( + OxTask.objects.get(id=r.id).status == OxTask.Status.SUCCESSFUL + for r in first + ) + ) + sessions = {n["session"] for n in read_notes(notes)} + assert len(sessions) == 1, "a persistent worker used more than one session" + (session,) = sessions + end_session(connection, session) + after = [(report_session if queries else tick).enqueue(notes) for _ in range(5)] + output = finish(proc, log) + finally: + if proc.poll() is None: + proc.kill() + proc.wait(timeout=10) + stored = [OxTask.objects.get(id=r.id) for r in first + after] + assert {s.status for s in stored} == {OxTask.Status.SUCCESSFUL}, output + assert "Unhandled error" not in output + extra_attempts = sum(s.attempts - 1 for s in stored) + # With a query, the first task to touch the dead session fails it and the + # retry runs on a new one. Without, the outcome write finds it and goes + # again on a new connection, and no attempt is spent. + assert extra_attempts == (1 if queries else 0), output + found = execution_threads(counts) + assert found["MainThread"] == 1 + # The session that was kept, and the one that replaced it. + assert_opened(found["execution"], 2, output) + assert all(s.lease_epoch == s.attempts for s in stored) + + +@ends_a_connection +def test_each_default_attempt_has_a_session_of_its_own(tmp_path): + """ + Control for the above: at the default nothing is kept between tasks, so + there is no idle session for the server to end and no state to leak from + one task to the next. + """ + notes = str(tmp_path / "notes.jsonl") + for _ in range(6): + report_session.enqueue(notes) + counts, output = run_batch(tmp_path, "default") + all_succeeded_once() + assert len({n["session"] for n in read_notes(notes)}) == 6, output + assert_opened(sum(execution_threads(counts).values()), 7, output) + + +def test_workers_with_and_without_persistence_share_a_queue(tmp_path): + """ + One worker keeps its connection and one reconnects for every attempt, + both on one queue. Every task completes, each attempt runs once, and the + lease fencing is as for any two workers: one claim, one epoch. + """ + notes = str(tmp_path / "notes.jsonl") + total = 40 + for i in range(total): + tick.enqueue(notes, 0.05, index=i) + kept, kept_counts = project(tmp_path, "persistent", "kept") + plain, plain_counts = project(tmp_path, "default", "plain") + flags = ("--batch", "--concurrency", "1") + kept_proc, kept_log = start(kept, tmp_path, *flags, name="kept") + plain_proc, plain_log = start(plain, tmp_path, *flags, name="plain") + kept_out = finish(kept_proc, kept_log) + plain_out = finish(plain_proc, plain_log) + + rows = list(OxTask.objects.all()) + assert len(rows) == total + assert {r.status for r in rows} == {OxTask.Status.SUCCESSFUL}, kept_out + plain_out + assert {(r.attempts, r.lease_epoch, len(r.worker_ids)) for r in rows} == {(1, 1, 1)} + ran = read_notes(notes) + assert sorted(n["index"] for n in ran) == list(range(total)) + by_pid = Counter(n["pid"] for n in ran) + assert set(by_pid) == {kept_proc.pid, plain_proc.pid}, "a worker never ran a task" + # The one that reconnects opens a connection per attempt plus its poll + # loop's; the one that keeps its connection opens two whatever it ran. + plain_connects = sum(connection_counter.by_thread(plain_counts).values()) + assert_opened(plain_connects, by_pid[plain_proc.pid] + 1, plain_out) + kept_connects = sum(connection_counter.by_thread(kept_counts).values()) + assert_opened(kept_connects, 2, kept_out)