feat!: add InboxCleaner and drop the Inbox cleanup methods

AGENTS.md

@@ -22,6 +22,8 @@ The things it does not say:
 
 - `Inbox` never commits, rolls back, begins, or opens a session or connection. Transaction
   ownership stays with the caller. Any new method must keep that.
+- `InboxCleaner` is the only code that begins and commits, and only on connections it opens from
+  the caller's engine. It never takes a caller's session and never disposes the engine.
 - `Inbox.claim` requires an open transaction on its session. That is what turns a wrongly wired
   second session into an error instead of a marker that never commits.
 - Concurrency claims are proven only by tests on separate connections, never one shared test

CONTEXT.md

@@ -36,3 +36,7 @@ _Avoid_: TTL (nothing expires on its own; cleanup deletes)
 **Cutoff**:
 The database timestamp, taken once per cleanup run, before which markers are deleted.
 _Avoid_: deadline, expiry time
+
+**Cleanup run**:
+One pass that takes a cutoff and deletes the markers older than it, batch by batch.
+_Avoid_: vacuum (Postgres VACUUM is something else), purge, sweep, expiry

README.md

@@ -7,8 +7,9 @@
 transaction as the consumer's own writes. A redelivered message finds its marker and is skipped. A
 failed transaction rolls the marker back, so a retry processes the message again.
 
-It works with SQLAlchemy's asyncio extension on PostgreSQL. It stores no payload, runs no worker,
-and never begins, commits or rolls back a transaction for you.
+It works with SQLAlchemy's asyncio extension on PostgreSQL. It stores no payload and runs no worker.
+`Inbox` never begins, commits or rolls back a transaction for you. Only `InboxCleaner` does that, on
+its own connections.
 
 ```bash
 pip install sqlalchemy-inbox
@@ -92,31 +93,38 @@ or fall back to Kafka offsets.
 
 ## Cleaning up
 
-Markers are kept until you delete them. Take one cutoff per cleanup run from the database clock,
-then delete in batches, one transaction per batch, until a batch deletes nothing:
+Markers are kept until you delete them. Run `InboxCleaner` from a cron job:
 
 ```python
-@dataclasses.dataclass(kw_only=True, frozen=True, slots=True)
-class CleanupInboxUseCase:
-    transaction: Transaction
-    inbox: Inbox
+import asyncio
+import datetime
 
-    async def __call__(self, *, retention: datetime.timedelta, batch_size: int) -> None:
-        async with self.transaction:
-            cutoff = await self.inbox.cleanup_cutoff(retention=retention)
-        while await self._delete_batch(cutoff=cutoff, limit=batch_size):
-            pass
+from sqlalchemy.ext.asyncio import create_async_engine
+from sqlalchemy_inbox import InboxCleaner
 
-    @postgres_retry
-    async def _delete_batch(self, *, cutoff: datetime.datetime, limit: int) -> int:
-        async with self.transaction:
-            deleted = await self.inbox.delete_expired_batch(cutoff=cutoff, limit=limit)
-            await self.transaction.commit()
-        return deleted
+
+async def main() -> None:
+    engine = create_async_engine(DATABASE_DSN)
+    try:
+        cleaner = InboxCleaner(engine=engine, table=INBOX_TABLE, retention=datetime.timedelta(days=7))
+        await cleaner.cleanup()
+    finally:
+        await engine.dispose()
+
+
+asyncio.run(main())
 ```
 
-Batches that committed stay deleted if a later batch fails, and the error reaches your scheduler.
-Cleanup skips markers that another transaction holds, so it never waits on a consumer.
+A cleanup run takes one cutoff from the database clock, `now()` minus `retention`, and deletes the
+markers older than it in batches of `batch_size` (1000 by default), one transaction per batch, until
+a batch deletes nothing. `cleanup()` returns a `CleanupResult` with the `cutoff` and the
+`deleted_count`. It logs each batch at DEBUG and the total at INFO to the `sqlalchemy_inbox` logger.
+
+Batches that committed stay deleted if a later batch fails or the task is cancelled, and the error
+reaches your cron job. The cleaner doesn't retry: the next run deletes what is left. It skips markers
+that another transaction holds, so it never waits on a consumer, and two runs at once share the
+work. Each batch runs at READ COMMITTED whatever the engine's default. The engine stays yours to
+dispose.
 
 ## Guarantees
 

docs/adr/0001-postgres-only.md

@@ -0,0 +1,13 @@
+# PostgreSQL only
+
+The package supports PostgreSQL and no other database. Its value is the guarantees in the README, and
+those come from how Postgres behaves: `INSERT ... ON CONFLICT DO NOTHING` waits on an uncommitted
+duplicate under READ COMMITTED, and cleanup relies on `FOR UPDATE SKIP LOCKED` and `DELETE ...
+RETURNING`. Each extra dialect would need its own code paths, its own guarantees and its own CI matrix.
+
+## Considered Options
+
+- SQLite as well, for tests without a server. The insert syntax matches, but SQLite has one writer at a
+  time and no row locks, so it would test a different concurrency model from the one in production.
+- MySQL. It has no `RETURNING`, rejects `LIMIT` in an `IN` subquery, and its `INSERT IGNORE` turns
+  unrelated errors into warnings, so both claim and cleanup would need separate implementations.

sqlalchemy_inbox/__init__.py

@@ -1,5 +1,6 @@
+from sqlalchemy_inbox.cleaner import CleanupResult, InboxCleaner
 from sqlalchemy_inbox.inbox import Inbox, SessionOrConnection
 from sqlalchemy_inbox.schema import make_inbox_table
 
 
-__all__ = ["Inbox", "SessionOrConnection", "make_inbox_table"]
+__all__ = ["CleanupResult", "Inbox", "InboxCleaner", "SessionOrConnection", "make_inbox_table"]

sqlalchemy_inbox/cleaner.py

@@ -0,0 +1,78 @@
+import contextlib
+import dataclasses
+import datetime as dt
+import logging
+from collections.abc import AsyncGenerator
+
+from sqlalchemy import Table, delete, func, select, tuple_
+from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine
+
+from sqlalchemy_inbox.schema import validate_table_identifiers
+
+
+_LOGGER = logging.getLogger(__name__)
+
+
+@dataclasses.dataclass(kw_only=True, frozen=True, slots=True)
+class CleanupResult:
+    cutoff: dt.datetime
+    deleted_count: int
+
+
+class InboxCleaner:
+    """Deletes markers older than *retention*, one transaction per batch.
+
+    Opens its own connections from *engine* but never disposes it: the caller owns the engine.
+    """
+
+    def __init__(self, *, engine: AsyncEngine, table: Table, retention: dt.timedelta, batch_size: int = 1000) -> None:
+        if not isinstance(engine, AsyncEngine):
+            msg = "InboxCleaner requires an sqlalchemy.ext.asyncio.AsyncEngine"
+            raise TypeError(msg)
+        validate_table_identifiers(table.name)
+        if retention <= dt.timedelta(0):
+            msg = f"retention must be positive (got {retention})"
+            raise ValueError(msg)
+        if batch_size < 1:
+            msg = f"batch_size must be positive (got {batch_size})"
+            raise ValueError(msg)
+        self._engine = engine
+        self._table = table
+        self._retention = retention
+        self._batch_size = batch_size
+
+    async def cleanup(self) -> CleanupResult:
+        """Take one cutoff from the database clock, then delete the markers older than it until a batch deletes nothing.
+
+        Each batch commits on its own and skips markers other transactions hold. If a batch fails or the task
+        is cancelled, that batch rolls back, earlier batches stay deleted, and the error propagates.
+        """
+        async with self._transaction() as connection:
+            cutoff = (await connection.execute(select(func.now() - self._retention))).scalar_one()
+        deleted_count = 0
+        while deleted := await self._delete_batch(cutoff):
+            deleted_count += deleted
+            _LOGGER.debug("Inbox cleanup batch deleted %d markers (%d so far)", deleted, deleted_count)
+        _LOGGER.info("Inbox cleanup deleted %d markers older than %s", deleted_count, cutoff.isoformat())
+        return CleanupResult(cutoff=cutoff, deleted_count=deleted_count)
+
+    async def _delete_batch(self, cutoff: dt.datetime) -> int:
+        key = tuple_(self._table.c.scope, self._table.c.message_id)
+        expired = (
+            select(self._table.c.scope, self._table.c.message_id)
+            .where(self._table.c.processed_at < cutoff)
+            .limit(self._batch_size)
+            .with_for_update(skip_locked=True)
+        )
+        async with self._transaction() as connection:
+            result = await connection.execute(
+                delete(self._table).where(key.in_(expired)).returning(self._table.c.scope)
+            )
+            return len(result.all())
+
+    @contextlib.asynccontextmanager
+    async def _transaction(self) -> AsyncGenerator[AsyncConnection]:
+        async with self._engine.connect() as connection:
+            await connection.execution_options(isolation_level="READ COMMITTED")
+            async with connection.begin():
+                yield connection

sqlalchemy_inbox/inbox.py

@@ -1,6 +1,4 @@
-import datetime as dt
-
-from sqlalchemy import Table, delete, func, select, tuple_
+from sqlalchemy import Table
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.ext.asyncio import AsyncConnection, AsyncSession
 
@@ -52,34 +50,6 @@ async def claim(self, *, scope: str, message_id: str) -> bool:
         result = await self._session.execute(statement)
         return result.first() is not None
 
-    async def cleanup_cutoff(self, *, retention: dt.timedelta) -> dt.datetime:
-        """Return the database's ``now()`` minus *retention*: take it once per cleanup run."""
-        if retention <= dt.timedelta(0):
-            msg = f"retention must be positive (got {retention})"
-            raise ValueError(msg)
-        self._require_transaction("cleanup_cutoff")
-        result = await self._session.execute(select(func.now() - retention))
-        return result.scalar_one()
-
-    async def delete_expired_batch(self, *, cutoff: dt.datetime, limit: int) -> int:
-        """Delete up to *limit* markers older than *cutoff*, skipping rows other transactions hold.
-
-        Returns the number deleted; a run is done when a batch returns 0.
-        """
-        if limit < 1:
-            msg = f"limit must be positive (got {limit})"
-            raise ValueError(msg)
-        self._require_transaction("delete_expired_batch")
-        key = tuple_(self._table.c.scope, self._table.c.message_id)
-        expired = (
-            select(self._table.c.scope, self._table.c.message_id)
-            .where(self._table.c.processed_at < cutoff)
-            .limit(limit)
-            .with_for_update(skip_locked=True)
-        )
-        result = await self._session.execute(delete(self._table).where(key.in_(expired)).returning(self._table.c.scope))
-        return len(result.all())
-
     def _require_transaction(self, method: str) -> None:
         if not self._session.in_transaction():
             msg = f"Inbox.{method} requires an open transaction on its session"