Fix a problem where tests were leaving leftover jobs in the database (#28)

Fix a problem where tests were leaving leftover jobs in the database
which turns out to be because SQLAlchemy stupidly doesn't start a
transaction when `begin()` is called, which was having the effect of the
unique job check starting a top-level transaction instead of savepoint,
thereby committing insert results outside of the test transactions.

This took an insanely long time to track down, and I ended up adding a
number of extra measures to help find the problem which I'm also keeping
for the time begin:

* `RIVER_DEBUG` activates debug logging for SQLAlchemy, showing queries
  and connection pool activity.

* An autorunning fixture checks for extra jobs after each test case to
  make sure nothing gets leftover.

* `SimpleArgs` starts to internalize the name of the test that created
  it, so upon seeing an extraneous row, it's easy to know exactly where
  it came from. It becomes a fixture to make this terse.

Author: brandur

README.md

@@ -97,7 +97,7 @@ insert_res.unique_skipped_as_duplicated
 
 ### Custom advisory lock prefix
 
-Unique job insertion takes a Postgres advisory lock to make sure that it's uniqueness check still works even if two conflicting insert operations are occurring in parallel. Postgres advisory locks share a global 64-bit namespace, which is a large enough space that it's unlikely for two advisory locks to ever conflict, but to _guarantee_ that River's advisory locks never interfere with an application's, River can be configured with a 32-bit advisory lock prefix which it will use for all its locks:
+Unique job insertion takes a Postgres advisory lock to make sure that its uniqueness check still works even if two conflicting insert operations are occurring in parallel. Postgres advisory locks share a global 64-bit namespace, which is a large enough space that it's unlikely for two advisory locks to ever conflict, but to _guarantee_ that River's advisory locks never interfere with an application's, River can be configured with a 32-bit advisory lock prefix which it will use for all its locks:
 
 ```python
 client = riverqueue.Client(riversqlalchemy.Driver(engine), advisory_lock_prefix: 123456)

src/riverqueue/driver/riversqlalchemy/dbsqlc/river_job.py

@@ -4,14 +4,20 @@
 # source: river_job.sql
 import dataclasses
 import datetime
-from typing import Any, List, Optional
+from typing import Any, AsyncIterator, Iterator, List, Optional
 
 import sqlalchemy
 import sqlalchemy.ext.asyncio
 
 from . import models
 
 
+JOB_GET_ALL = """-- name: job_get_all \\:many
+SELECT id, args, attempt, attempted_at, attempted_by, created_at, errors, finalized_at, kind, max_attempts, metadata, priority, queue, state, scheduled_at, tags
+FROM river_job
+"""
+
+
 JOB_GET_BY_KIND_AND_UNIQUE_PROPERTIES = """-- name: job_get_by_kind_and_unique_properties \\:one
 SELECT id, args, attempt, attempted_at, attempted_by, created_at, errors, finalized_at, kind, max_attempts, metadata, priority, queue, state, scheduled_at, tags
 FROM river_job
@@ -125,6 +131,28 @@ class Querier:
     def __init__(self, conn: sqlalchemy.engine.Connection):
         self._conn = conn
 
+    def job_get_all(self) -> Iterator[models.RiverJob]:
+        result = self._conn.execute(sqlalchemy.text(JOB_GET_ALL))
+        for row in result:
+            yield models.RiverJob(
+                id=row[0],
+                args=row[1],
+                attempt=row[2],
+                attempted_at=row[3],
+                attempted_by=row[4],
+                created_at=row[5],
+                errors=row[6],
+                finalized_at=row[7],
+                kind=row[8],
+                max_attempts=row[9],
+                metadata=row[10],
+                priority=row[11],
+                queue=row[12],
+                state=row[13],
+                scheduled_at=row[14],
+                tags=row[15],
+            )
+
     def job_get_by_kind_and_unique_properties(self, arg: JobGetByKindAndUniquePropertiesParams) -> Optional[models.RiverJob]:
         row = self._conn.execute(sqlalchemy.text(JOB_GET_BY_KIND_AND_UNIQUE_PROPERTIES), {
             "p1": arg.kind,
@@ -213,6 +241,28 @@ class AsyncQuerier:
     def __init__(self, conn: sqlalchemy.ext.asyncio.AsyncConnection):
         self._conn = conn
 
+    async def job_get_all(self) -> AsyncIterator[models.RiverJob]:
+        result = await self._conn.stream(sqlalchemy.text(JOB_GET_ALL))
+        async for row in result:
+            yield models.RiverJob(
+                id=row[0],
+                args=row[1],
+                attempt=row[2],
+                attempted_at=row[3],
+                attempted_by=row[4],
+                created_at=row[5],
+                errors=row[6],
+                finalized_at=row[7],
+                kind=row[8],
+                max_attempts=row[9],
+                metadata=row[10],
+                priority=row[11],
+                queue=row[12],
+                state=row[13],
+                scheduled_at=row[14],
+                tags=row[15],
+            )
+
     async def job_get_by_kind_and_unique_properties(self, arg: JobGetByKindAndUniquePropertiesParams) -> Optional[models.RiverJob]:
         row = (await self._conn.execute(sqlalchemy.text(JOB_GET_BY_KIND_AND_UNIQUE_PROPERTIES), {
             "p1": arg.kind,

src/riverqueue/driver/riversqlalchemy/dbsqlc/river_job.sql

@@ -35,6 +35,10 @@ CREATE TABLE river_job(
     CONSTRAINT kind_length CHECK (char_length(kind) > 0 AND char_length(kind) < 128)
 );
 
+-- name: JobGetAll :many
+SELECT *
+FROM river_job;
+
 -- name: JobGetByKindAndUniqueProperties :one
 SELECT *
 FROM river_job

tests/client_test.py

@@ -8,8 +8,6 @@
 from riverqueue.driver import DriverProtocol, ExecutorProtocol
 import sqlalchemy
 
-from tests.simple_args import SimpleArgs
-
 
 @pytest.fixture
 def mock_driver() -> DriverProtocol:
@@ -38,17 +36,17 @@ def client(mock_driver) -> Client:
     return Client(mock_driver)
 
 
-def test_insert_with_only_args(client, mock_exec):
+def test_insert_with_only_args(client, mock_exec, simple_args):
     mock_exec.job_get_by_kind_and_unique_properties.return_value = None
     mock_exec.job_insert.return_value = "job_row"
 
-    insert_res = client.insert(SimpleArgs())
+    insert_res = client.insert(simple_args)
 
     mock_exec.job_insert.assert_called_once()
     assert insert_res.job == "job_row"
 
 
-def test_insert_tx(mock_driver, client):
+def test_insert_tx(mock_driver, client, simple_args):
     mock_exec = MagicMock(spec=ExecutorProtocol)
     mock_exec.job_get_by_kind_and_unique_properties.return_value = None
     mock_exec.job_insert.return_value = "job_row"
@@ -61,17 +59,17 @@ def mock_unwrap_executor(tx: sqlalchemy.Transaction):
 
     mock_driver.unwrap_executor.side_effect = mock_unwrap_executor
 
-    insert_res = client.insert_tx(mock_tx, SimpleArgs())
+    insert_res = client.insert_tx(mock_tx, simple_args)
 
     mock_exec.job_insert.assert_called_once()
     assert insert_res.job == "job_row"
 
 
-def test_insert_with_insert_opts_from_args(client, mock_exec):
+def test_insert_with_insert_opts_from_args(client, mock_exec, simple_args):
     mock_exec.job_insert.return_value = "job_row"
 
     insert_res = client.insert(
-        SimpleArgs(),
+        simple_args,
         insert_opts=InsertOpts(
             max_attempts=23, priority=2, queue="job_custom_queue", tags=["job_custom"]
         ),
@@ -121,7 +119,7 @@ def to_json() -> str:
     assert insert_args.tags == ["job_custom"]
 
 
-def test_insert_with_insert_opts_precedence(client, mock_exec):
+def test_insert_with_insert_opts_precedence(client, mock_exec, simple_args):
     @dataclass
     class MyArgs:
         kind = "my_args"
@@ -142,7 +140,7 @@ def to_json() -> str:
     mock_exec.job_insert.return_value = "job_row"
 
     insert_res = client.insert(
-        SimpleArgs(),
+        simple_args,
         insert_opts=InsertOpts(
             max_attempts=17, priority=3, queue="my_queue", tags=["custom"]
         ),
@@ -158,13 +156,13 @@ def to_json() -> str:
     assert insert_args.tags == ["custom"]
 
 
-def test_insert_with_unique_opts_by_args(client, mock_exec):
+def test_insert_with_unique_opts_by_args(client, mock_exec, simple_args):
     insert_opts = InsertOpts(unique_opts=UniqueOpts(by_args=True))
 
     mock_exec.job_get_by_kind_and_unique_properties.return_value = None
     mock_exec.job_insert.return_value = "job_row"
 
-    insert_res = client.insert(SimpleArgs(), insert_opts=insert_opts)
+    insert_res = client.insert(simple_args, insert_opts=insert_opts)
 
     mock_exec.job_insert.assert_called_once()
     assert insert_res.job == "job_row"
@@ -175,15 +173,17 @@ def test_insert_with_unique_opts_by_args(client, mock_exec):
 
 
 @patch("datetime.datetime")
-def test_insert_with_unique_opts_by_period(mock_datetime, client, mock_exec):
+def test_insert_with_unique_opts_by_period(
+    mock_datetime, client, mock_exec, simple_args
+):
     mock_datetime.now.return_value = datetime(2024, 6, 1, 12, 0, 0, tzinfo=timezone.utc)
 
     insert_opts = InsertOpts(unique_opts=UniqueOpts(by_period=900))
 
     mock_exec.job_get_by_kind_and_unique_properties.return_value = None
     mock_exec.job_insert.return_value = "job_row"
 
-    insert_res = client.insert(SimpleArgs(), insert_opts=insert_opts)
+    insert_res = client.insert(simple_args, insert_opts=insert_opts)
 
     mock_exec.job_insert.assert_called_once()
     assert insert_res.job == "job_row"
@@ -193,13 +193,13 @@ def test_insert_with_unique_opts_by_period(mock_datetime, client, mock_exec):
     assert call_args.kind == "simple"
 
 
-def test_insert_with_unique_opts_by_queue(client, mock_exec):
+def test_insert_with_unique_opts_by_queue(client, mock_exec, simple_args):
     insert_opts = InsertOpts(unique_opts=UniqueOpts(by_queue=True))
 
     mock_exec.job_get_by_kind_and_unique_properties.return_value = None
     mock_exec.job_insert.return_value = "job_row"
 
-    insert_res = client.insert(SimpleArgs(), insert_opts=insert_opts)
+    insert_res = client.insert(simple_args, insert_opts=insert_opts)
 
     mock_exec.job_insert.assert_called_once()
     assert insert_res.job == "job_row"
@@ -209,13 +209,13 @@ def test_insert_with_unique_opts_by_queue(client, mock_exec):
     assert call_args.kind == "simple"
 
 
-def test_insert_with_unique_opts_by_state(client, mock_exec):
+def test_insert_with_unique_opts_by_state(client, mock_exec, simple_args):
     insert_opts = InsertOpts(unique_opts=UniqueOpts(by_state=["available", "running"]))
 
     mock_exec.job_get_by_kind_and_unique_properties.return_value = None
     mock_exec.job_insert.return_value = "job_row"
 
-    insert_res = client.insert(SimpleArgs(), insert_opts=insert_opts)
+    insert_res = client.insert(simple_args, insert_opts=insert_opts)
 
     mock_exec.job_insert.assert_called_once()
     assert insert_res.job == "job_row"
@@ -259,20 +259,20 @@ def to_json() -> None:
     assert "args should return non-nil from `to_json`" == str(ex.value)
 
 
-def test_tag_validation(client):
+def test_tag_validation(client, simple_args):
     client.insert(
-        SimpleArgs(), insert_opts=InsertOpts(tags=["foo", "bar", "baz", "foo-bar-baz"])
+        simple_args, insert_opts=InsertOpts(tags=["foo", "bar", "baz", "foo-bar-baz"])
     )
 
     with pytest.raises(AssertionError) as ex:
-        client.insert(SimpleArgs(), insert_opts=InsertOpts(tags=["commas,bad"]))
+        client.insert(simple_args, insert_opts=InsertOpts(tags=["commas,bad"]))
     assert (
         r"tags should be less than 255 characters in length and match regex \A[\w][\w\-]+[\w]\Z"
         == str(ex.value)
     )
 
     with pytest.raises(AssertionError) as ex:
-        client.insert(SimpleArgs(), insert_opts=InsertOpts(tags=["a" * 256]))
+        client.insert(simple_args, insert_opts=InsertOpts(tags=["a" * 256]))
     assert (
         r"tags should be less than 255 characters in length and match regex \A[\w][\w\-]+[\w]\Z"
         == str(ex.value)

tests/conftest.py

@@ -1,13 +1,30 @@
+from dataclasses import dataclass
+import json
 import os
+from typing import Iterator
 
 import pytest
 import sqlalchemy
 import sqlalchemy.ext.asyncio
 
+from riverqueue.driver.riversqlalchemy.dbsqlc import river_job
+
+
+def engine_opts() -> dict:
+    """
+    Use to pass verbose logging options to an SQLAlchemy when `RIVER_DEBUG=true`
+    in the environment.
+    """
+
+    if os.getenv("RIVER_DEBUG") == "true":
+        return dict(echo=True, echo_pool="debug")
+
+    return dict()
+
 
 @pytest.fixture(scope="session")
 def engine() -> sqlalchemy.Engine:
-    return sqlalchemy.create_engine(test_database_url())
+    return sqlalchemy.create_engine(test_database_url(), **engine_opts())
 
 
 # @pytest_asyncio.fixture(scope="session")
@@ -22,10 +39,17 @@ def engine_async() -> sqlalchemy.ext.asyncio.AsyncEngine:
         # This statement disables pooling which isn't ideal, but I've spent
         # too many hours trying to figure this out so I'm calling it.
         poolclass=sqlalchemy.pool.NullPool,
+        **engine_opts(),
     )
 
 
 def test_database_url(is_async: bool = False) -> str:
+    """
+    Produces a test URL based on `TEST_DATABASE_URL` or River's default
+    convention and modifies it so that it's protocol includes an appropriate
+    driver to make SQLAlchemy happy.
+    """
+
     database_url = os.getenv("TEST_DATABASE_URL", "postgres://localhost/river_test")
 
     # sqlalchemy removed support for postgres:// for reasons beyond comprehension
@@ -36,3 +60,44 @@ def test_database_url(is_async: bool = False) -> str:
         database_url = database_url.replace("postgresql://", "postgresql+asyncpg://")
 
     return database_url
+
+
+@dataclass
+class SimpleArgs:
+    test_name: str
+
+    kind: str = "simple"
+
+    def to_json(self) -> str:
+        return json.dumps({"test_name": self.test_name})
+
+
+@pytest.fixture
+def simple_args(request: pytest.FixtureRequest):
+    """
+    Returns an instance of SimpleArgs encapsulating the running test's name. This
+    can be useful in cases where a test is accidentally leaving leftovers in the
+    database.
+    """
+
+    return SimpleArgs(test_name=request.node.name)
+
+
+@pytest.fixture(autouse=True)
+def check_leftover_jobs(engine) -> Iterator[None]:
+    """
+    Autorunning fixture that checks for leftover jobs after each test case. I
+    previously had a huge amount of trouble tracking down tests that were
+    inserting rows despite being in a test transaction and ended up adding this
+    check, along with naming inserted jobs after their test case. If it turns
+    these measures haven't been needed in a long time, we can probably remove
+    them.
+    """
+
+    yield
+
+    with engine.begin() as conn_tx:
+        jobs = river_job.Querier(conn_tx).job_get_all()
+        assert (
+            list(jobs) == []
+        ), "test case should not have persisted any jobs after run"