test: run suite in parallel with pytest-xdist

AGENTS.md

@@ -36,9 +36,12 @@ uv run ruff format . --check  # Format check
 uv run lint-imports         # Hexagonal architecture boundary validation
 uv run mypy .               # Type checking (strict, targets Python 3.10)
 
-# Run all tests
+# Run all tests (parallel by default: 4 xdist workers, one Postgres container each)
 uv run pytest
 
+# Run tests serially (disable pytest-xdist)
+uv run pytest -n 0
+
 # Run a single test file
 uv run pytest test/test_queries.py
 
@@ -225,7 +228,9 @@ When a comment explains a non-obvious workaround, keep it short and put it adjac
 
 ## Testing Conventions
 
-- **pytest config**: `asyncio_mode = "auto"`, 20s timeout per test, `--durations=15`
+- **pytest config**: `asyncio_mode = "auto"`, 20s timeout per test, `--durations=15`, `-n 4` (pytest-xdist)
+- Tests run in parallel via **pytest-xdist**: session-scoped fixtures are per-worker, so each worker starts its own Postgres container and template database. Pass `-n 0` to run serially, or `-n <N>` to change worker count.
+- Keep tests xdist-safe: no fixed ports, no shared files, no cross-test state. Database isolation comes from the per-test `CREATE DATABASE ... TEMPLATE` fixture.
 - Declare async tests as `async def test_...` -- no `@pytest.mark.asyncio` decorator needed
 - Annotate all test function parameters: `async def test_foo(apgdriver: db.Driver, N: int) -> None:`
 - Use `@pytest.mark.parametrize` for multi-value testing

pyproject.toml

@@ -89,6 +89,7 @@ dev = [
     "opentelemetry-semantic-conventions>=0.41b0",
     "pytest-mock>=3.14.0",
     "pytest-timeout>=2.3.1",
+    "pytest-xdist>=3.8.0",
     "ruff",
     "tqdm",
     "types-croniter",
@@ -154,7 +155,9 @@ explicit_package_bases = true
 asyncio_mode = "auto"
 timeout = "20"
 timeout_func_only = true
-addopts = "--durations=15"
+# -n 4: modest parallelism; each xdist worker gets its own session-scoped
+# Postgres container, so this scales the suite across a few extra servers.
+addopts = "--durations=15 -n 4"
 
 # ---------------------------------------------------------------------------
 # Import-linter: enforce hexagonal architecture boundaries

test/conftest.py

@@ -21,7 +21,7 @@
     uvloop = None  # type: ignore[assignment]
 
 from testcontainers.core.container import DockerContainer
-from testcontainers.core.wait_strategies import LogMessageWaitStrategy
+from testcontainers.core.wait_strategies import HealthcheckWaitStrategy
 
 from pgqueuer.db import SyncPsycopgDriver
 
@@ -62,7 +62,21 @@ def __init__(
         self.with_env("POSTGRES_USER", self.username)
         self.with_env("POSTGRES_PASSWORD", self.password)
         self.with_env("POSTGRES_DB", self.dbname)
-        self.waiting_for(LogMessageWaitStrategy("database system is ready to accept connections"))
+        # Readiness comes from docker's own healthcheck (as in ci.yml service
+        # containers), probing pg_isready over TCP: the image's initdb
+        # bootstrap server listens only on the unix socket and logs a
+        # misleading "ready to accept connections" line, so the container
+        # cannot report healthy before the real server accepts TCP.
+        self._kwargs.setdefault(
+            "healthcheck",
+            {
+                "test": ["CMD", "pg_isready", "-h", "127.0.0.1", "-U", self.username],
+                "interval": 500_000_000,  # docker durations are nanoseconds
+                "timeout": 2_000_000_000,
+                "retries": 120,
+            },
+        )
+        self.waiting_for(HealthcheckWaitStrategy())
 
     def get_connection_url(self, host: str | None = None) -> str:
         if self._container is None:
@@ -94,6 +108,10 @@ async def postgres_container() -> AsyncGenerator[str, None]:
 
     # https://postgresqlco.nf/doc/en/param/vacuum_buffer_usage_limit/
     # Assume worst case for backwards compatibility
+    #
+    # Memory is bounded per container (shared_buffers, max_wal_size, tmpfs size)
+    # because each xdist worker starts its own container and the data dir lives
+    # on tmpfs (RAM).
     commands = [
         "postgres",
         "-c",
@@ -107,19 +125,22 @@ async def postgres_container() -> AsyncGenerator[str, None]:
         "-c",
         "checkpoint_timeout=30min",
         "-c",
-        "max_wal_size=4GB",
+        "max_wal_size=512MB",
         "-c",
         "checkpoint_completion_target=0.9",
         "-c",
-        "shared_buffers=256MB",
+        "shared_buffers=128MB",
         "-c",
         "autovacuum_naptime=5s",
     ] + (["-c", "vacuum_buffer_usage_limit=8MB"] if int(postgres_version) >= 16 else [])
 
     container = (
-        PgQueuerPostgresContainer(f"postgres:{postgres_version}", driver=None)
+        PgQueuerPostgresContainer(
+            f"postgres:{postgres_version}",
+            driver=None,
+            tmpfs={"/var/lib/pg/data": "rw,size=1g"},
+        )
         .with_command(commands)
-        .with_kwargs(tmpfs={"/var/lib/pg/data": "rw"})
         .with_envs(PGDATA="/var/lib/pg/data")
     )