diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..aaba8ff --- /dev/null +++ b/.dockerignore @@ -0,0 +1,14 @@ +.env +.coverage +.gitignore +.idea +.ruff_cache +.vscode +.git +.pytest_cache +.DS_Store +*.yml +Dockerfile +**/__pycache__ +.hypothesis +.venv diff --git a/.github/workflows/_checks.yml b/.github/workflows/_checks.yml new file mode 100644 index 0000000..181cc63 --- /dev/null +++ b/.github/workflows/_checks.yml @@ -0,0 +1,124 @@ +name: checks + +on: + workflow_call: {} + +jobs: + lint: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v6 + - uses: extractions/setup-just@v4 + - uses: astral-sh/setup-uv@v8.2.0 + with: + enable-cache: true + cache-dependency-glob: "**/pyproject.toml" + - run: uv python install 3.11 + - run: uv python pin 3.11 + - run: just install lint-ci + + links: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v6 + - name: Check local links + uses: lycheeverse/lychee-action@v2 + with: + args: >- + --offline + --no-progress + --remap + 'https://github\.com/${{ github.repository }}/(?:blob|tree)/main/(.*) file://${{ github.workspace }}/$1' + '**/*.md' + + pytest: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: + - "3.11" + - "3.12" + - "3.13" + - "3.14" + - "3.14t" + - "3.15" + - "3.15t" + services: + postgres: + image: postgres:17 + env: + POSTGRES_USER: inbox + POSTGRES_PASSWORD: inbox + POSTGRES_DB: inbox + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U inbox" + --health-interval 10s + --health-timeout 5s + --health-retries 5 + steps: + - uses: actions/checkout@v6 + - uses: extractions/setup-just@v4 + - uses: astral-sh/setup-uv@v8.2.0 + with: + enable-cache: true + cache-dependency-glob: "**/pyproject.toml" + - run: uv python install ${{ matrix.python-version }} + - run: uv python pin ${{ matrix.python-version }} + - run: just install + - name: Assert the GIL is actually disabled + if: endsWith(matrix.python-version, 't') + run: uv run --no-sync python -c "import sys, asyncpg, sqlalchemy, sqlalchemy_inbox; assert not sys._is_gil_enabled(), 'GIL re-enabled - a dependency lacks free-threaded support'" + - run: just test-ci + env: + PYTHONDONTWRITEBYTECODE: 1 + PYTHONUNBUFFERED: 1 + POSTGRES_DSN: postgresql+asyncpg://inbox:inbox@127.0.0.1:5432/inbox + + floors: + # Installs the declared floors on each Python, against the oldest supported PostgreSQL. + if: github.event_name != 'schedule' + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: + - "3.11" + - "3.12" + - "3.13" + - "3.14" + services: + postgres: + image: postgres:14 + env: + POSTGRES_USER: inbox + POSTGRES_PASSWORD: inbox + POSTGRES_DB: inbox + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U inbox" + --health-interval 10s + --health-timeout 5s + --health-retries 5 + steps: + - uses: actions/checkout@v6 + - uses: astral-sh/setup-uv@v8.2.0 + with: + enable-cache: true + cache-dependency-glob: "**/pyproject.toml" + - run: uv python install ${{ matrix.python-version }} + - run: uv python pin ${{ matrix.python-version }} + - run: uv venv + - if: matrix.python-version == '3.11' + run: uv pip compile pyproject.toml --no-deps --resolution lowest-direct -o floors.txt + - if: matrix.python-version != '3.11' + run: uv pip compile pyproject.toml --resolution lowest-direct --no-build -o floors.txt + - run: uv pip install -r pyproject.toml --group dev --constraints floors.txt --no-build + - run: uv run --no-sync pytest . --no-cov + env: + PYTHONDONTWRITEBYTECODE: 1 + PYTHONUNBUFFERED: 1 + POSTGRES_DSN: postgresql+asyncpg://inbox:inbox@127.0.0.1:5432/inbox diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..751045b --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,15 @@ +name: main + +on: + push: + branches: + - main + pull_request: {} + +concurrency: + group: ${{ github.head_ref || github.run_id }} + cancel-in-progress: true + +jobs: + checks: + uses: ./.github/workflows/_checks.yml diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml new file mode 100644 index 0000000..ec20b10 --- /dev/null +++ b/.github/workflows/release.yml @@ -0,0 +1,55 @@ +name: Release + +# Tag-driven: pushing a semver tag publishes to PyPI and creates the matching +# GitHub Release. Replaces the old `on: release: published` publish.yml — that +# trigger is removed so the published Release this workflow creates can't re-fire +# it (double-publish). The tag is the sole entry point; by convention a tag is +# only cut off a green main, so there is no in-workflow CI gate. +on: + push: + tags: + - '[0-9]+.[0-9]+.[0-9]+' # stable: 2.7.2 + - '[0-9]+.[0-9]+.[0-9]+[a-z]+[0-9]+' # pre-release: 2.0.0rc1, 4.0.0a2 + +# contents: write -> create the GitHub Release; id-token: write -> OIDC for PyPI Trusted Publishing. +permissions: + contents: write + id-token: write + +jobs: + release: + runs-on: ubuntu-latest + environment: pypi # scopes the PyPI Trusted Publisher; hook for approval rules + steps: + - uses: actions/checkout@v6 + - uses: extractions/setup-just@v4 + - uses: astral-sh/setup-uv@v7 + + # PyPI is irreversible, so it runs FIRST: if it fails the job stops and no + # GitHub Release is created advertising a version that never reached PyPI. + # `just publish` derives the version from $GITHUB_REF_NAME (the tag name). + # Auth via PyPI Trusted Publishing (OIDC); no PYPI_TOKEN. Needs a Trusted + # Publisher on the sqlalchemy-inbox PyPI project (env: pypi, workflow: release.yml). + - run: just publish + + # The Release body is GitHub's generated notes, rendered from the squashed + # PR titles since the previous tag — so a conventional-commit title is what + # a reader gets. A release wanting prose is edited after the fact with + # `gh release edit --notes-file`. A tag with a letter (2.0.0rc1) is a + # pre-release -> flagged so GitHub won't mark it "Latest". + - name: Resolve release metadata + id: meta + run: | + set -euo pipefail + if [[ "$GITHUB_REF_NAME" =~ [a-z] ]]; then + echo "prerelease=true" >> "$GITHUB_OUTPUT" + else + echo "prerelease=false" >> "$GITHUB_OUTPUT" + fi + + - name: Publish GitHub Release + uses: softprops/action-gh-release@v3 + with: + generate_release_notes: true + prerelease: ${{ steps.meta.outputs.prerelease }} + draft: false diff --git a/.github/workflows/scheduled.yml b/.github/workflows/scheduled.yml new file mode 100644 index 0000000..4bf659e --- /dev/null +++ b/.github/workflows/scheduled.yml @@ -0,0 +1,20 @@ +name: scheduled-dep-check +on: + schedule: + - cron: "0 6 * * *" # Daily 06:00 UTC + workflow_dispatch: {} + +concurrency: + group: scheduled-dep-check + cancel-in-progress: false + +jobs: + checks: + uses: ./.github/workflows/_checks.yml + + report-failure: + needs: checks + if: failure() && github.event_name == 'schedule' + permissions: + issues: write + uses: modern-python/.github/.github/workflows/report-scheduled-failure.yml@main diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..b9d607b --- /dev/null +++ b/.gitignore @@ -0,0 +1,22 @@ +# Generic things +*.pyc +*~ +__pycache__/* +*.swp +*.sqlite3 +*.map +.vscode +.idea +.DS_Store +.env +.pytest_cache +.ruff_cache +.coverage +htmlcov/ +coverage.xml +pytest.xml +dist/ +.python-version +.venv +uv.lock +/site/ diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..42cc593 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,39 @@ +# AGENTS.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## Project Overview + +`sqlalchemy-inbox` is a transactional inbox ledger for SQLAlchemy asyncio on PostgreSQL. +[`CONTEXT.md`](CONTEXT.md) opens with what it does and owns the vocabulary. Read it before naming a +concept in code, a test name, or an issue title. + +## Commands + +`just` (task runner) and `uv` (package manager). The [`justfile`](justfile) is the source of truth. +The things it does not say: + +- `just test [args]` forwards args **unquoted**, so a spaced `-k` expression word-splits and fails. +- Every test needs PostgreSQL and skips if `POSTGRES_DSN` is unreachable. The compose Postgres + publishes no host port, so run tests through `just test`. The coverage gate is on in CI; pass + `--no-cov` while iterating locally if you add `--cov`. + +## Architecture + +- `Inbox` never commits, rolls back, begins, or opens a session or connection. Transaction + ownership stays with the caller. Any new method must keep that. +- `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 + transaction. + +## Workflow + +Every link in `README.md` must be absolute: `https://github.com/modern-python/sqlalchemy-inbox/blob/main/`. +`README.md` is also the PyPI long description, and PyPI does not rewrite relative links. + +## Code Style + +- **Never use local/inline imports**, tests included. `ruff` catches them. +- Annotate every function argument. +- Docstrings: public API documents the contract; internal helpers get a one-line contract at most. diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..43c994c --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1 @@ +@AGENTS.md diff --git a/CONTEXT.md b/CONTEXT.md new file mode 100644 index 0000000..8357d64 --- /dev/null +++ b/CONTEXT.md @@ -0,0 +1,38 @@ +# sqlalchemy-inbox + +A processed-message ledger for consumers on SQLAlchemy's asyncio extension and PostgreSQL: one row +per (scope, message id), inserted in the caller's transaction so it commits or rolls back with the +business writes. Postgres-only. It stores no payload and runs no worker. + +## Language + +A term is listed only when there is a synonym to reject, or a meaning subtle enough that code and +docs must agree on it. General programming vocabulary does not belong here, however heavily this +project uses it. + +**Marker**: +One ledger row, keyed by `(scope, message_id)`. Its presence after commit means the database +effects for that key have committed. +_Avoid_: lock, processed event, dedup record + +**Scope**: +A stable string naming one logical consumer of a message, such as `chats.create-chat.v1`. Two +consumers that process the same message independently use different scopes. +_Avoid_: handler name, consumer group (both change on refactors and deployment renames) + +**Message id**: +The producer-assigned identity of a message, the same on every delivery. +_Avoid_: offset, delivery id (both change between deliveries) + +**Claim**: +Inserting a marker in the caller's open transaction. A claim that inserted returns `True`; one +that found a committed marker returns `False`. +_Avoid_: acquire, lock, register + +**Retention**: +How long a marker is kept. After it is deleted, the same key is new again. +_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 diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..4caca67 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,24 @@ +FROM python:3.14-slim + +COPY --from=ghcr.io/astral-sh/uv:latest /uv /bin/uv +RUN useradd --no-create-home --gid root runner + +ENV UV_PROJECT_ENVIRONMENT=/code/.venv \ + UV_NO_MANAGED_PYTHON=1 \ + UV_NO_CACHE=true \ + UV_LINK_MODE=copy + +WORKDIR /code + +COPY pyproject.toml . +COPY uv.lock . + +RUN uv sync --all-extras --frozen --no-install-project + +COPY . . + +RUN chown -R runner:root /code && chmod -R g=u /code + +USER runner + +RUN uv sync --all-extras --frozen diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..667cfff --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Artur Shiriev + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 0000000..d721076 --- /dev/null +++ b/README.md @@ -0,0 +1,182 @@ +# sqlalchemy-inbox + +[![CI](https://github.com/modern-python/sqlalchemy-inbox/actions/workflows/ci.yml/badge.svg)](https://github.com/modern-python/sqlalchemy-inbox/actions/workflows/ci.yml) +[![License](https://img.shields.io/github/license/modern-python/sqlalchemy-inbox.svg)](https://github.com/modern-python/sqlalchemy-inbox/blob/main/LICENSE) + +`sqlalchemy-inbox` records which messages a consumer has processed, in the same PostgreSQL +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. + +Work in progress: the package is not released yet. + +## The table + +Attach the table to your own `MetaData` and write the Alembic migration yourself. The package owns +no migrations. + +```python +from sqlalchemy import MetaData +from sqlalchemy_inbox import make_inbox_table + +metadata = MetaData() +INBOX_TABLE = make_inbox_table(metadata, table_name="inbox") +``` + +The table has three columns: `scope` and `message_id` (both `varchar(255)`, together the primary +key) and `processed_at` (`timestamptz`, defaulting to the database's `now()`), with an index on +`processed_at` for cleanup. + +## Registering with modern-di + +`Inbox` takes a session and the table, so it registers like a repository. The session provider must +be REQUEST-scoped and cached, so that `Inbox` and your `Transaction` get the same session object: + +```python +from db_retry import Transaction +from modern_di import Group, Scope, providers +from sqlalchemy_inbox import Inbox + + +class Database(Group): + session = providers.Factory( + scope=Scope.REQUEST, + creator=create_session, + cache=providers.CacheSettings(finalizer=close_session), + kwargs={"engine": database_engine}, + ) + transaction = providers.Factory(scope=Scope.REQUEST, creator=Transaction, kwargs={"session": session}) + + +class Repositories(Group, scope=Scope.REQUEST): + inbox = providers.Factory(creator=Inbox, kwargs={"session": Database.session, "table": INBOX_TABLE}) +``` + +If a service wires an uncached session by mistake, `Inbox` gets a second session that has no open +transaction, and `claim` raises `RuntimeError` instead of writing a marker that would never commit. + +## Processing a message + +Claim the message inside your transaction, before the business writes: + +```python +@dataclasses.dataclass(kw_only=True, frozen=True, slots=True) +class CreateChatUseCase: + transaction: Transaction + inbox: Inbox + chats_repository: ChatsRepository + + @postgres_retry + async def __call__(self, *, command: CreateChat) -> None: + async with self.transaction: + if not await self.inbox.claim(scope="chats.create-chat.v1", message_id=command.idempotency_id): + return + await self.chats_repository.create(command) + await self.transaction.commit() +``` + +`claim` returns `True` when this transaction inserted the marker, and `False` when the message is +already committed. If another transaction holds an uncommitted marker for the same key, `claim` +waits for it to finish. Retrying the whole block with `postgres_retry` is safe: if an earlier attempt +committed, the claim on the retry returns `False`. + +`claim` does not accept `None`. If a message can arrive without an id, decide in the handler what to +do with it: reject it, or skip the inbox in a branch you can see. The package does not hash payloads +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: + +```python +@dataclasses.dataclass(kw_only=True, frozen=True, slots=True) +class CleanupInboxUseCase: + transaction: Transaction + inbox: Inbox + + 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 + + @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 +``` + +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. + +## Guarantees + +What the package guarantees, given a PostgreSQL transaction that you own: + +- The marker and your writes commit together or not at all. That includes a failure in your code, + a database error, a cancelled task and a failed `COMMIT`. +- Two transactions processing the same `(scope, message_id)` at once commit the database effects at + most once. The second waits for the first and returns `False` if the first committed, `True` if it + rolled back. +- A redelivery after a successful commit returns `False`, including when the process died after the + commit but before the broker was acknowledged. +- If you can't tell whether a commit landed, run the transaction again: `claim` returns `False` if it + did and `True` if it didn't. + +What it does not guarantee: + +- Exactly-once delivery from end to end. The guarantee covers database effects for one key while + its marker is retained. +- Effects outside PostgreSQL. An HTTP call made in a transaction that later rolls back has already + happened, and the retry makes it again. Send those through a transactional outbox, or make the + receiving side idempotent. +- Anything after the marker is deleted. Once cleanup removes it, the same key counts as a new + message. +- Under REPEATABLE READ or SERIALIZABLE, a conflicting `claim` raises a serialization failure + instead of returning `False`. db-retry's `postgres_retry` retries it, and the retry returns + `False`. + +## Choosing a scope + +A scope names one logical consumer of a message. Write it as a constant, such as +`"chats.create-chat.v1"`. + +Don't build it from a function name, module path, consumer group or topic. Those change when code +is refactored or a deployment is renamed, and a changed scope makes every retained message look new. + +Two consumers that each need to process the same message use different scopes. Change the version +suffix only when you want old messages to be processed again. + +## Choosing a retention + +Retention must cover the longest time a message can come back. That includes: + +- the Kafka topic's retention, which defaults to 7 days, +- DLQ retention plus the time it takes to replay from it, +- manual replays and backfills. + +A day of retention is enough only if nothing can be redelivered after a day. Each marker is dated by +the start of the transaction that wrote it, so a marker from a long transaction expires that much +earlier. At retention measured in days, that makes no practical difference. + +## Kafka offsets for failed handlers + +The inbox makes redelivery safe. It cannot cause a redelivery. With +[faststream-concurrent-aiokafka](https://github.com/modern-python/faststream-concurrent-aiokafka), +a handler that raises still has its offset committed. Only a cancelled handler holds the offset +back. So when your transaction fails, the marker rolls back as it should, but Kafka will not send +the message again. + +If a failed message has to be processed later, retry inside the handler (for example with +`postgres_retry`) or send it to a dead-letter queue before the handler returns. The same applies to +errors db-retry deliberately doesn't retry, such as `StatementCompletionUnknownError`, where you +can't tell whether the commit landed. + +## Links + +- [License](https://github.com/modern-python/sqlalchemy-inbox/blob/main/LICENSE) diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..d1cc101 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,29 @@ +services: + application: + build: + context: . + dockerfile: ./Dockerfile + restart: always + volumes: + - .:/code + - /code/.venv + depends_on: + postgres: + condition: service_healthy + environment: + - POSTGRES_DSN=postgresql+asyncpg://inbox:inbox@postgres:5432/inbox + stdin_open: true + tty: true + + postgres: + image: postgres:17 + restart: always + environment: + - POSTGRES_USER=inbox + - POSTGRES_PASSWORD=inbox + - POSTGRES_DB=inbox + healthcheck: + test: ["CMD-SHELL", "pg_isready -U inbox -d inbox"] + interval: 1s + timeout: 3s + retries: 30 diff --git a/justfile b/justfile new file mode 100644 index 0000000..c999526 --- /dev/null +++ b/justfile @@ -0,0 +1,42 @@ +default: install lint build test + +down: + docker compose down --remove-orphans --volumes + +sh: + docker compose run --service-ports application bash + +test *args: down && down + docker compose run application uv run pytest {{ args }} + +test-ci: + uv run --no-sync pytest --cov=. --cov-report term-missing --cov-report xml + +test-branch: + @just test --cov=. --cov-branch + +build: + docker compose build application + +install: + uv lock --upgrade + uv sync --all-extras --all-groups --frozen + +lint: + uv run eof-fixer . + uv run ruff format + uv run ruff check --fix + uv run ty check + +lint-ci: + uv run eof-fixer . --check + uv run ruff format --check + uv run ruff check --no-fix + uv run ty check + +# Auth via PyPI Trusted Publishing (OIDC); uv publish auto-detects the CI id-token. +publish: + rm -rf dist + uv version $GITHUB_REF_NAME + uv build + uv publish diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..941c169 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,100 @@ +[project] +name = "sqlalchemy-inbox" +version = "0" +description = "Transactional inbox ledger for SQLAlchemy asyncio and PostgreSQL: dedupe consumed messages in the caller's transaction" +authors = [ + { name = "Artur Shiriev", email = "me@shiriev.ru" }, +] +readme = "README.md" +requires-python = ">=3.11,<4" +license = "MIT" +keywords = ["sqlalchemy", "inbox", "transactional-inbox", "idempotent-consumer", "postgresql", "asyncio", "python"] +classifiers = [ + "Development Status :: 3 - Alpha", + "Intended Audience :: Developers", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Programming Language :: Python :: 3.14", + "Programming Language :: Python :: 3.15", + "Programming Language :: Python :: Free Threading :: 2 - Beta", + "Typing :: Typed", + "Topic :: Software Development :: Libraries", +] +dependencies = [ + # 2.0.31 is the first release that tolerates the __static_attributes__ and __firstlineno__ + # attributes Python 3.13 added; below it SQLAlchemy's TypingOnly check fails at import. + # 2.0.20 is the lowest version the tests can verify: db-retry, a dev dependency, requires it. + "sqlalchemy[asyncio]>=2.0.20; python_version < '3.13'", + "sqlalchemy[asyncio]>=2.0.31; python_version == '3.13'", + # 2.0.54 is the first release whose cyextensions declare Py_MOD_GIL_NOT_USED. There is no marker + # for a free-threaded build, so ordinary 3.14 carries the same floor. + "sqlalchemy[asyncio]>=2.0.54; python_version >= '3.14'", +] + +[dependency-groups] +dev = [ + "pytest>=8", + "pytest-asyncio>=0.24", + "pytest-cov>=5", + "asyncpg>=0.29", + "db-retry>=0.9", +] +lint = [ + "ruff", + "ty", + "eof-fixer", +] + +[build-system] +requires = ["uv_build"] +build-backend = "uv_build" + +[tool.uv.build-backend] +module-name = "sqlalchemy_inbox" +module-root = "" + +[tool.ruff] +fix = true +unsafe-fixes = true +line-length = 120 + +[tool.ruff.lint] +select = ["ALL"] +ignore = [ + "D1", # docstrings are not forced; a docstring exists when it says something + "D203", # conflicts with D211 + "D213", # conflicts with D212 + "COM812", # conflicts with the formatter + "ISC001", # conflicts with the formatter + "CPY001", # no per-file copyright header + "FBT", # boolean positional arguments are fine + "TCH", # imports stay real; TYPE_CHECKING-only imports break runtime introspection +] +isort.lines-after-imports = 2 +isort.no-lines-before = ["standard-library", "local-folder"] + +[tool.ruff.lint.extend-per-file-ignores] +"tests/**" = [ + "S101", # allow asserts + "PLR2004", # allow magic values +] + +[tool.pytest.ini_options] +testpaths = ["tests"] +addopts = "" +asyncio_mode = "auto" +asyncio_default_fixture_loop_scope = "function" + +[project.urls] +Repository = "https://github.com/modern-python/sqlalchemy-inbox" +Issues = "https://github.com/modern-python/sqlalchemy-inbox/issues" +Changelog = "https://github.com/modern-python/sqlalchemy-inbox/releases" + +[tool.coverage.run] +# SQLAlchemy's async layer runs statements in greenlets; untracked switches drop executed lines. +concurrency = ["greenlet", "thread"] + +[tool.coverage.report] +fail_under = 100 +exclude_also = ["if typing.TYPE_CHECKING:"] diff --git a/sqlalchemy_inbox/__init__.py b/sqlalchemy_inbox/__init__.py new file mode 100644 index 0000000..40b7a2c --- /dev/null +++ b/sqlalchemy_inbox/__init__.py @@ -0,0 +1,5 @@ +from sqlalchemy_inbox.inbox import Inbox, SessionOrConnection +from sqlalchemy_inbox.schema import make_inbox_table + + +__all__ = ["Inbox", "SessionOrConnection", "make_inbox_table"] diff --git a/sqlalchemy_inbox/inbox.py b/sqlalchemy_inbox/inbox.py new file mode 100644 index 0000000..dbccbac --- /dev/null +++ b/sqlalchemy_inbox/inbox.py @@ -0,0 +1,86 @@ +import datetime as dt + +from sqlalchemy import Table, delete, func, select, tuple_ +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.ext.asyncio import AsyncConnection, AsyncSession + +from sqlalchemy_inbox.schema import MESSAGE_ID_MAX_LENGTH, SCOPE_MAX_LENGTH, validate_table_identifiers + + +SessionOrConnection = AsyncSession | AsyncConnection + + +def _validate_key_part(name: str, value: str, max_length: int) -> None: + if not value: + msg = f"{name} must not be empty" + raise ValueError(msg) + if len(value) > max_length: + msg = f"{name} must be at most {max_length} characters (got {len(value)})" + raise ValueError(msg) + + +class Inbox: + """Processed-message ledger bound to the caller's session. + + Never commits, rolls back, begins, or opens a session or connection: the caller owns the transaction. + """ + + def __init__(self, session: SessionOrConnection, table: Table) -> None: + if not isinstance(session, SessionOrConnection): + msg = "Inbox requires an sqlalchemy.ext.asyncio.AsyncSession or AsyncConnection" + raise TypeError(msg) + validate_table_identifiers(table.name) + self._session = session + self._table = table + + async def claim(self, *, scope: str, message_id: str) -> bool: + """Insert the marker for ``(scope, message_id)`` in the caller's open transaction. + + Returns ``True`` if this transaction inserted it and the caller should do the work, ``False`` if the + key is already committed. While another transaction holds an uncommitted marker for the key, waits for + it to finish. Raises ``RuntimeError`` if the session has no open transaction. + """ + _validate_key_part("scope", scope, SCOPE_MAX_LENGTH) + _validate_key_part("message_id", message_id, MESSAGE_ID_MAX_LENGTH) + self._require_transaction("claim") + statement = ( + pg_insert(self._table) + .values(scope=scope, message_id=message_id) + .on_conflict_do_nothing(index_elements=["scope", "message_id"]) + .returning(self._table.c.scope) + ) + 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" + raise RuntimeError(msg) diff --git a/sqlalchemy_inbox/py.typed b/sqlalchemy_inbox/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/sqlalchemy_inbox/schema.py b/sqlalchemy_inbox/schema.py new file mode 100644 index 0000000..98adcb6 --- /dev/null +++ b/sqlalchemy_inbox/schema.py @@ -0,0 +1,41 @@ +from sqlalchemy import Column, DateTime, Index, MetaData, PrimaryKeyConstraint, String, Table, func + + +SCOPE_MAX_LENGTH = 255 +MESSAGE_ID_MAX_LENGTH = 255 + +_POSTGRES_IDENT_MAX_BYTES = 63 +_PKEY_SUFFIX = "_pkey" +_PROCESSED_AT_IDX_SUFFIX = "_processed_at_idx" + + +def validate_table_identifiers(table_name: str) -> None: + """Raise ``ValueError`` if an identifier derived from *table_name* exceeds Postgres' 63-byte limit.""" + longest = max( + (f"{table_name}{suffix}".encode() for suffix in (_PKEY_SUFFIX, _PROCESSED_AT_IDX_SUFFIX)), + key=len, + ) + if len(longest) > _POSTGRES_IDENT_MAX_BYTES: + msg = ( + f"table_name {table_name!r} too long: the derived identifier {longest.decode()!r} " + f"must fit in {_POSTGRES_IDENT_MAX_BYTES} bytes (got {len(longest)})" + ) + raise ValueError(msg) + + +def make_inbox_table(metadata: MetaData, table_name: str = "inbox") -> Table: + """Build the inbox ``Table`` and attach it to *metadata*. + + The package does not own the schema: wire the table into your own ``MetaData`` and write the + Alembic migration yourself. + """ + validate_table_identifiers(table_name) + return Table( + table_name, + metadata, + Column("scope", String(SCOPE_MAX_LENGTH), nullable=False), + Column("message_id", String(MESSAGE_ID_MAX_LENGTH), nullable=False), + Column("processed_at", DateTime(timezone=True), nullable=False, server_default=func.now()), + PrimaryKeyConstraint("scope", "message_id", name=f"{table_name}{_PKEY_SUFFIX}"), + Index(f"{table_name}{_PROCESSED_AT_IDX_SUFFIX}", "processed_at"), + ) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..e34d85d --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,77 @@ +import asyncio +import dataclasses +import os +import uuid +from collections.abc import AsyncIterator + +import pytest +from sqlalchemy import BigInteger, Column, MetaData, String, Table, func, select, text +from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine + +from sqlalchemy_inbox import make_inbox_table + + +PG_DSN = os.environ.get("POSTGRES_DSN", "postgresql+asyncpg://inbox:inbox@localhost:5432/inbox") + + +@pytest.fixture +async def pg_engine() -> AsyncIterator[AsyncEngine]: + engine = create_async_engine(PG_DSN) + try: + async with engine.connect() as conn: + await conn.exec_driver_sql("SELECT 1") + except Exception as exc: # noqa: BLE001 # pragma: no cover - only when Postgres is unreachable; the test then skips + await engine.dispose() + pytest.skip(f"Postgres not available at {PG_DSN}: {exc}") + yield engine + await engine.dispose() + + +@dataclasses.dataclass(frozen=True, slots=True) +class Tables: + inbox: Table + business: Table + + +@pytest.fixture +async def tables(pg_engine: AsyncEngine) -> AsyncIterator[Tables]: + """Per-test inbox table plus a ``business`` table that stands in for a service's own writes.""" + metadata = MetaData() + suffix = uuid.uuid4().hex[:12] + business = Table( + f"test_business_{suffix}", + metadata, + Column("id", BigInteger, primary_key=True, autoincrement=True), + Column("message_id", String(255), nullable=False), + ) + created = Tables(inbox=make_inbox_table(metadata, table_name=f"test_inbox_{suffix}"), business=business) + async with pg_engine.begin() as conn: + await conn.run_sync(metadata.create_all) + yield created + async with pg_engine.begin() as conn: + await conn.run_sync(metadata.drop_all) + + +@pytest.fixture +def session_maker(pg_engine: AsyncEngine) -> async_sessionmaker[AsyncSession]: + return async_sessionmaker(pg_engine, expire_on_commit=False) + + +async def count_rows(engine: AsyncEngine, table: Table) -> int: + async with engine.connect() as conn: + return (await conn.execute(select(func.count()).select_from(table))).scalar_one() + + +async def wait_until_lock_wait(engine: AsyncEngine, pid: int) -> None: + async with engine.connect() as conn: + for _ in range(200): + waiting = ( + await conn.execute( + text("SELECT wait_event_type = 'Lock' FROM pg_stat_activity WHERE pid = :pid"), {"pid": pid} + ) + ).scalar_one() + if waiting: + return + await asyncio.sleep(0.025) # pragma: no cover - only when the first poll runs before the lock wait + msg = f"backend {pid} never waited on a lock" # pragma: no cover - only when the claim did not block + raise AssertionError(msg) # pragma: no cover - only when the claim did not block diff --git a/tests/test_claim.py b/tests/test_claim.py new file mode 100644 index 0000000..112ae9b --- /dev/null +++ b/tests/test_claim.py @@ -0,0 +1,126 @@ +import pytest +from sqlalchemy import MetaData, Table, insert +from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker + +from sqlalchemy_inbox import Inbox +from tests.conftest import Tables, count_rows + + +class _BusinessError(Exception): + pass + + +async def _process(session: AsyncSession, tables: Tables, *, scope: str, message_id: str) -> bool: + async with session.begin(): + if not await Inbox(session, tables.inbox).claim(scope=scope, message_id=message_id): + return False + await session.execute(insert(tables.business).values(message_id=message_id)) + return True + + +async def test_first_delivery_commits_marker_and_business_write( + pg_engine: AsyncEngine, tables: Tables, session_maker: async_sessionmaker[AsyncSession] +) -> None: + async with session_maker() as session: + assert await _process(session, tables, scope="s", message_id="m-1") + + assert await count_rows(pg_engine, tables.inbox) == 1 + assert await count_rows(pg_engine, tables.business) == 1 + + +async def test_sequential_redelivery_is_skipped( + pg_engine: AsyncEngine, tables: Tables, session_maker: async_sessionmaker[AsyncSession] +) -> None: + async with session_maker() as session: + assert await _process(session, tables, scope="s", message_id="m-1") + async with session_maker() as session: + assert not await _process(session, tables, scope="s", message_id="m-1") + + assert await count_rows(pg_engine, tables.inbox) == 1 + assert await count_rows(pg_engine, tables.business) == 1 + + +async def test_different_scopes_process_the_same_message_independently( + pg_engine: AsyncEngine, tables: Tables, session_maker: async_sessionmaker[AsyncSession] +) -> None: + async with session_maker() as session: + assert await _process(session, tables, scope="first", message_id="m-1") + async with session_maker() as session: + assert await _process(session, tables, scope="second", message_id="m-1") + + assert await count_rows(pg_engine, tables.inbox) == 2 + assert await count_rows(pg_engine, tables.business) == 2 + + +async def test_failure_after_claim_rolls_back_marker_and_business_write( + pg_engine: AsyncEngine, tables: Tables, session_maker: async_sessionmaker[AsyncSession] +) -> None: + async def fail_after_claim(session: AsyncSession) -> None: + async with session.begin(): + assert await Inbox(session, tables.inbox).claim(scope="s", message_id="m-1") + await session.execute(insert(tables.business).values(message_id="m-1")) + raise _BusinessError + + async with session_maker() as session: + with pytest.raises(_BusinessError): + await fail_after_claim(session) + + assert await count_rows(pg_engine, tables.inbox) == 0 + assert await count_rows(pg_engine, tables.business) == 0 + async with session_maker() as session: + assert await _process(session, tables, scope="s", message_id="m-1") + + +async def test_inbox_rejects_a_non_session(tables: Tables) -> None: + with pytest.raises(TypeError, match="AsyncSession or AsyncConnection"): + Inbox(object(), tables.inbox) # ty: ignore[invalid-argument-type] + + +def test_inbox_rejects_an_over_long_table_name() -> None: + with pytest.raises(ValueError, match="63 bytes"): + Inbox(AsyncSession(), Table("t" * 50, MetaData())) + + +async def test_claim_without_an_open_transaction_raises( + pg_engine: AsyncEngine, tables: Tables, session_maker: async_sessionmaker[AsyncSession] +) -> None: + async with session_maker() as session: + with pytest.raises(RuntimeError, match="open transaction"): + await Inbox(session, tables.inbox).claim(scope="s", message_id="m-1") + assert not session.in_transaction() + + assert await count_rows(pg_engine, tables.inbox) == 0 + + +@pytest.mark.parametrize( + ("scope", "message_id"), + [("", "m-1"), ("s", ""), ("s" * 256, "m-1"), ("s", "m" * 256)], +) +async def test_claim_rejects_empty_or_over_long_keys( + pg_engine: AsyncEngine, + tables: Tables, + session_maker: async_sessionmaker[AsyncSession], + scope: str, + message_id: str, +) -> None: + async with session_maker() as session, session.begin(): + with pytest.raises(ValueError, match=r"scope|message_id"): + await Inbox(session, tables.inbox).claim(scope=scope, message_id=message_id) + + assert await count_rows(pg_engine, tables.inbox) == 0 + + +async def test_claim_on_a_wrongly_wired_idle_session_raises( + pg_engine: AsyncEngine, tables: Tables, session_maker: async_sessionmaker[AsyncSession] +) -> None: + async def write_with_wrong_inbox_session(business_session: AsyncSession, inbox_session: AsyncSession) -> None: + async with business_session.begin(): + await business_session.execute(insert(tables.business).values(message_id="m-1")) + await Inbox(inbox_session, tables.inbox).claim(scope="s", message_id="m-1") + + async with session_maker() as business_session, session_maker() as inbox_session: + with pytest.raises(RuntimeError, match="open transaction"): + await write_with_wrong_inbox_session(business_session, inbox_session) + + assert await count_rows(pg_engine, tables.inbox) == 0 + assert await count_rows(pg_engine, tables.business) == 0 diff --git a/tests/test_cleanup.py b/tests/test_cleanup.py new file mode 100644 index 0000000..f6dc28f --- /dev/null +++ b/tests/test_cleanup.py @@ -0,0 +1,204 @@ +import asyncio +import datetime as dt +import uuid + +import pytest +from sqlalchemy import MetaData, Table, func, insert, select, text +from sqlalchemy.exc import DBAPIError +from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker + +from sqlalchemy_inbox import Inbox, make_inbox_table +from tests.conftest import Tables, wait_until_lock_wait + + +HOUR = dt.timedelta(hours=1) + + +class _BatchError(Exception): + pass + + +async def _add_marker(engine: AsyncEngine, table: Table, message_id: str, age: dt.timedelta) -> None: + async with engine.begin() as conn: + await conn.execute(insert(table).values(scope="s", message_id=message_id, processed_at=func.now() - age)) + + +async def _message_ids(engine: AsyncEngine, table: Table) -> list[str]: + async with engine.connect() as conn: + return list((await conn.execute(select(table.c.message_id).order_by(table.c.message_id))).scalars()) + + +async def _cutoff( + session_maker: async_sessionmaker[AsyncSession], table: Table, retention: dt.timedelta +) -> dt.datetime: + async with session_maker() as session, session.begin(): + return await Inbox(session, table).cleanup_cutoff(retention=retention) + + +async def _delete_batch( + session_maker: async_sessionmaker[AsyncSession], table: Table, cutoff: dt.datetime, limit: int +) -> int: + async with session_maker() as session, session.begin(): + return await Inbox(session, table).delete_expired_batch(cutoff=cutoff, limit=limit) + + +async def test_cutoff_comes_from_the_database_clock( + tables: Tables, session_maker: async_sessionmaker[AsyncSession] +) -> None: + async with session_maker() as session, session.begin(): + cutoff = await Inbox(session, tables.inbox).cleanup_cutoff(retention=HOUR) + database_now = (await session.execute(select(func.now()))).scalar_one() + + assert cutoff == database_now - HOUR + + +async def test_only_markers_older_than_the_cutoff_are_deleted( + pg_engine: AsyncEngine, tables: Tables, session_maker: async_sessionmaker[AsyncSession] +) -> None: + await _add_marker(pg_engine, tables.inbox, "old", 2 * HOUR) + await _add_marker(pg_engine, tables.inbox, "recent", dt.timedelta(minutes=10)) + cutoff = await _cutoff(session_maker, tables.inbox, HOUR) + + assert await _delete_batch(session_maker, tables.inbox, cutoff, 100) == 1 + assert await _message_ids(pg_engine, tables.inbox) == ["recent"] + + +async def test_each_batch_respects_its_limit( + pg_engine: AsyncEngine, tables: Tables, session_maker: async_sessionmaker[AsyncSession] +) -> None: + for index in range(5): + await _add_marker(pg_engine, tables.inbox, f"m-{index}", 2 * HOUR) + cutoff = await _cutoff(session_maker, tables.inbox, HOUR) + + deleted = [await _delete_batch(session_maker, tables.inbox, cutoff, 2) for _ in range(4)] + + assert deleted == [2, 2, 1, 0] + + +async def test_a_run_keeps_its_cutoff_for_markers_added_later( + pg_engine: AsyncEngine, tables: Tables, session_maker: async_sessionmaker[AsyncSession] +) -> None: + await _add_marker(pg_engine, tables.inbox, "old", 2 * HOUR) + cutoff = await _cutoff(session_maker, tables.inbox, HOUR) + async with pg_engine.begin() as conn: + await conn.execute( + insert(tables.inbox).values( + scope="s", message_id="after-cutoff", processed_at=cutoff + dt.timedelta(seconds=1) + ) + ) + + assert await _delete_batch(session_maker, tables.inbox, cutoff, 100) == 1 + assert await _message_ids(pg_engine, tables.inbox) == ["after-cutoff"] + + +async def test_committed_batches_survive_a_failed_batch( + pg_engine: AsyncEngine, tables: Tables, session_maker: async_sessionmaker[AsyncSession] +) -> None: + for index in range(4): + await _add_marker(pg_engine, tables.inbox, f"m-{index}", 2 * HOUR) + cutoff = await _cutoff(session_maker, tables.inbox, HOUR) + + assert await _delete_batch(session_maker, tables.inbox, cutoff, 2) == 2 + + async def fail_after_delete(session: AsyncSession) -> None: + async with session.begin(): + assert await Inbox(session, tables.inbox).delete_expired_batch(cutoff=cutoff, limit=2) == 2 + raise _BatchError + + async with session_maker() as session: + with pytest.raises(_BatchError): + await fail_after_delete(session) + + assert len(await _message_ids(pg_engine, tables.inbox)) == 2 + assert await _delete_batch(session_maker, tables.inbox, cutoff, 2) == 2 + assert await _message_ids(pg_engine, tables.inbox) == [] + + +async def test_cleanup_skips_markers_locked_by_another_transaction( + pg_engine: AsyncEngine, tables: Tables, session_maker: async_sessionmaker[AsyncSession] +) -> None: + await _add_marker(pg_engine, tables.inbox, "locked", 2 * HOUR) + await _add_marker(pg_engine, tables.inbox, "free", 2 * HOUR) + cutoff = await _cutoff(session_maker, tables.inbox, HOUR) + + async with session_maker() as holder, holder.begin(): + await holder.execute(select(tables.inbox).where(tables.inbox.c.message_id == "locked").with_for_update()) + deleted = await asyncio.wait_for(_delete_batch(session_maker, tables.inbox, cutoff, 100), timeout=5) + + assert deleted == 1 + assert await _message_ids(pg_engine, tables.inbox) == ["locked"] + + +async def test_cleanup_errors_reach_the_caller( + session_maker: async_sessionmaker[AsyncSession], +) -> None: + missing = make_inbox_table(MetaData(), table_name=f"missing_{uuid.uuid4().hex[:12]}") + async with session_maker() as session: + with pytest.raises(DBAPIError): + async with session.begin(): + await Inbox(session, missing).delete_expired_batch(cutoff=dt.datetime.now(tz=dt.UTC), limit=10) + + +async def test_an_expired_message_is_processed_again( + pg_engine: AsyncEngine, tables: Tables, session_maker: async_sessionmaker[AsyncSession] +) -> None: + await _add_marker(pg_engine, tables.inbox, "m-1", 2 * HOUR) + async with session_maker() as session, session.begin(): + assert not await Inbox(session, tables.inbox).claim(scope="s", message_id="m-1") + + cutoff = await _cutoff(session_maker, tables.inbox, HOUR) + assert await _delete_batch(session_maker, tables.inbox, cutoff, 100) == 1 + + async with session_maker() as session, session.begin(): + assert await Inbox(session, tables.inbox).claim(scope="s", message_id="m-1") + + +async def test_claim_during_uncommitted_cleanup_waits_then_claims( + pg_engine: AsyncEngine, tables: Tables, session_maker: async_sessionmaker[AsyncSession] +) -> None: + await _add_marker(pg_engine, tables.inbox, "m-1", 2 * HOUR) + cutoff = await _cutoff(session_maker, tables.inbox, HOUR) + + async with session_maker() as cleaner, session_maker() as consumer: + await cleaner.begin() + assert await Inbox(cleaner, tables.inbox).delete_expired_batch(cutoff=cutoff, limit=100) == 1 + + await consumer.begin() + pid = (await consumer.execute(text("SELECT pg_backend_pid()"))).scalar_one() + competing = asyncio.create_task(Inbox(consumer, tables.inbox).claim(scope="s", message_id="m-1")) + await wait_until_lock_wait(pg_engine, pid) + assert not competing.done() + + await cleaner.commit() + assert await competing + await consumer.commit() + + assert await _message_ids(pg_engine, tables.inbox) == ["m-1"] + + +async def test_cutoff_rejects_a_non_positive_retention( + tables: Tables, session_maker: async_sessionmaker[AsyncSession] +) -> None: + async with session_maker() as session, session.begin(): + with pytest.raises(ValueError, match="retention"): + await Inbox(session, tables.inbox).cleanup_cutoff(retention=dt.timedelta(0)) + + +async def test_delete_rejects_a_non_positive_limit( + tables: Tables, session_maker: async_sessionmaker[AsyncSession] +) -> None: + async with session_maker() as session, session.begin(): + with pytest.raises(ValueError, match="limit"): + await Inbox(session, tables.inbox).delete_expired_batch(cutoff=dt.datetime.now(tz=dt.UTC), limit=0) + + +async def test_cleanup_without_an_open_transaction_raises( + tables: Tables, session_maker: async_sessionmaker[AsyncSession] +) -> None: + async with session_maker() as session: + inbox = Inbox(session, tables.inbox) + with pytest.raises(RuntimeError, match="open transaction"): + await inbox.cleanup_cutoff(retention=HOUR) + with pytest.raises(RuntimeError, match="open transaction"): + await inbox.delete_expired_batch(cutoff=dt.datetime.now(tz=dt.UTC), limit=10) + assert not session.in_transaction() diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py new file mode 100644 index 0000000..804ffd7 --- /dev/null +++ b/tests/test_concurrency.py @@ -0,0 +1,205 @@ +import asyncio +import uuid + +import pytest +from db_retry import Transaction, postgres_retry +from db_retry.retriable import is_retriable +from sqlalchemy import BigInteger, Column, MetaData, String, Table, UniqueConstraint, insert, text +from sqlalchemy.exc import DBAPIError, IntegrityError +from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker + +from sqlalchemy_inbox import Inbox, make_inbox_table +from tests.conftest import Tables, count_rows, wait_until_lock_wait + + +class _AckLostError(Exception): + pass + + +async def _backend_pid(session: AsyncSession) -> int: + return (await session.execute(text("SELECT pg_backend_pid()"))).scalar_one() + + +async def _claim_and_write(session: AsyncSession, tables: Tables, message_id: str) -> bool: + async with session.begin(): + if not await Inbox(session, tables.inbox).claim(scope="s", message_id=message_id): + return False + await session.execute(insert(tables.business).values(message_id=message_id)) + return True + + +async def _blocked_claim_after( + engine: AsyncEngine, + tables: Tables, + session_maker: async_sessionmaker[AsyncSession], + *, + first_commits: bool, +) -> bool: + async with session_maker() as first, session_maker() as second: + await first.begin() + assert await Inbox(first, tables.inbox).claim(scope="s", message_id="m-1") + await first.execute(insert(tables.business).values(message_id="m-1")) + + await second.begin() + pid = await _backend_pid(second) + competing = asyncio.create_task(Inbox(second, tables.inbox).claim(scope="s", message_id="m-1")) + await wait_until_lock_wait(engine, pid) + assert not competing.done() + + if first_commits: + await first.commit() + else: + await first.rollback() + acquired = await competing + if acquired: + await second.execute(insert(tables.business).values(message_id="m-1")) + await second.commit() + return acquired + + +async def test_competing_claim_waits_and_skips_when_first_commits( + pg_engine: AsyncEngine, tables: Tables, session_maker: async_sessionmaker[AsyncSession] +) -> None: + assert not await _blocked_claim_after(pg_engine, tables, session_maker, first_commits=True) + + assert await count_rows(pg_engine, tables.inbox) == 1 + assert await count_rows(pg_engine, tables.business) == 1 + + +async def test_competing_claim_proceeds_when_first_rolls_back( + pg_engine: AsyncEngine, tables: Tables, session_maker: async_sessionmaker[AsyncSession] +) -> None: + assert await _blocked_claim_after(pg_engine, tables, session_maker, first_commits=False) + + assert await count_rows(pg_engine, tables.inbox) == 1 + assert await count_rows(pg_engine, tables.business) == 1 + + +async def test_many_concurrent_deliveries_commit_one_effect( + pg_engine: AsyncEngine, tables: Tables, session_maker: async_sessionmaker[AsyncSession] +) -> None: + async def deliver() -> bool: + async with session_maker() as session: + return await _claim_and_write(session, tables, "m-1") + + results = await asyncio.gather(*(deliver() for _ in range(20))) + + assert results.count(True) == 1 + assert await count_rows(pg_engine, tables.inbox) == 1 + assert await count_rows(pg_engine, tables.business) == 1 + + +async def test_repeatable_read_conflict_is_a_retriable_serialization_failure( + pg_engine: AsyncEngine, tables: Tables, session_maker: async_sessionmaker[AsyncSession] +) -> None: + attempts: list[bool] = [] + + @postgres_retry(retries=2) + async def deliver() -> bool: + async with Transaction(session=session_maker(), isolation_level="REPEATABLE READ") as transaction: + await transaction.session.execute(text("SELECT 1")) + if not attempts: + async with session_maker() as other: + assert await _claim_and_write(other, tables, "m-1") + attempts.append(True) + return await Inbox(transaction.session, tables.inbox).claim(scope="s", message_id="m-1") + + async with Transaction(session=session_maker(), isolation_level="REPEATABLE READ") as transaction: + await transaction.session.execute(text("SELECT 1")) + async with session_maker() as other: + assert await _claim_and_write(other, tables, "m-0") + with pytest.raises(DBAPIError) as raised: + await Inbox(transaction.session, tables.inbox).claim(scope="s", message_id="m-0") + assert is_retriable(raised.value) + + assert not await deliver() + assert len(attempts) == 2 + assert await count_rows(pg_engine, tables.business) == 2 + + +async def test_cancelled_delivery_rolls_back_and_can_be_retried( + pg_engine: AsyncEngine, tables: Tables, session_maker: async_sessionmaker[AsyncSession] +) -> None: + claimed = asyncio.Event() + + async def deliver() -> None: + async with Transaction(session=session_maker()) as transaction: + assert await Inbox(transaction.session, tables.inbox).claim(scope="s", message_id="m-1") + await transaction.session.execute(insert(tables.business).values(message_id="m-1")) + claimed.set() + await asyncio.Event().wait() + + task = asyncio.create_task(deliver()) + await claimed.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + assert await count_rows(pg_engine, tables.inbox) == 0 + assert await count_rows(pg_engine, tables.business) == 0 + async with session_maker() as session: + assert await _claim_and_write(session, tables, "m-1") + + +async def test_failed_commit_persists_nothing_and_retry_redoes_the_work( + pg_engine: AsyncEngine, session_maker: async_sessionmaker[AsyncSession] +) -> None: + metadata = MetaData() + suffix = uuid.uuid4().hex[:12] + inbox_table = make_inbox_table(metadata, table_name=f"test_inbox_{suffix}") + business = Table( + f"test_deferred_{suffix}", + metadata, + Column("id", BigInteger, primary_key=True, autoincrement=True), + Column("value", String(255), nullable=False), + UniqueConstraint("value", deferrable=True, initially="DEFERRED"), + ) + async with pg_engine.begin() as conn: + await conn.run_sync(metadata.create_all) + await conn.execute(insert(business).values(value="taken")) + try: + + async def commit_a_duplicate(session: AsyncSession) -> None: + async with session.begin(): + assert await Inbox(session, inbox_table).claim(scope="s", message_id="m-1") + await session.execute(insert(business).values(value="taken")) + + async with session_maker() as session: + with pytest.raises(IntegrityError): + await commit_a_duplicate(session) + + assert await count_rows(pg_engine, inbox_table) == 0 + async with session_maker() as session, session.begin(): + assert await Inbox(session, inbox_table).claim(scope="s", message_id="m-1") + await session.execute(insert(business).values(value="free")) + assert await count_rows(pg_engine, inbox_table) == 1 + assert await count_rows(pg_engine, business) == 2 + finally: + async with pg_engine.begin() as conn: + await conn.run_sync(metadata.drop_all) + + +async def test_redelivery_after_commit_with_lost_ack_is_skipped( + pg_engine: AsyncEngine, tables: Tables, session_maker: async_sessionmaker[AsyncSession] +) -> None: + async def deliver() -> bool: + async with session_maker() as session: + acquired = await _claim_and_write(session, tables, "m-1") + if acquired: + raise _AckLostError + return acquired + + with pytest.raises(_AckLostError): + await deliver() + assert not await deliver() + + assert await count_rows(pg_engine, tables.inbox) == 1 + assert await count_rows(pg_engine, tables.business) == 1 + + +async def test_claim_works_on_an_async_connection(pg_engine: AsyncEngine, tables: Tables) -> None: + async with pg_engine.connect() as conn: + async with conn.begin(): + assert await Inbox(conn, tables.inbox).claim(scope="s", message_id="m-1") + async with conn.begin(): + assert not await Inbox(conn, tables.inbox).claim(scope="s", message_id="m-1") diff --git a/tests/test_schema.py b/tests/test_schema.py new file mode 100644 index 0000000..c121e32 --- /dev/null +++ b/tests/test_schema.py @@ -0,0 +1,56 @@ +import uuid + +import pytest +from sqlalchemy import MetaData, text +from sqlalchemy.ext.asyncio import AsyncEngine + +from sqlalchemy_inbox import make_inbox_table + + +async def test_table_has_scope_message_id_primary_key_and_processed_at_index(pg_engine: AsyncEngine) -> None: + metadata = MetaData() + table_name = f"test_inbox_{uuid.uuid4().hex[:12]}" + make_inbox_table(metadata, table_name=table_name) + async with pg_engine.begin() as conn: + await conn.run_sync(metadata.create_all) + try: + async with pg_engine.connect() as conn: + primary_key = ( + await conn.execute( + text( + "SELECT array_agg(a.attname ORDER BY k.ord) FROM pg_index i " + "CROSS JOIN LATERAL unnest(i.indkey) WITH ORDINALITY AS k(attnum, ord) " + "JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = k.attnum " + "WHERE i.indrelid = CAST(:t AS regclass) AND i.indisprimary" + ), + {"t": table_name}, + ) + ).scalar_one() + processed_at_default = ( + await conn.execute( + text( + "SELECT pg_get_expr(d.adbin, d.adrelid) FROM pg_attrdef d " + "JOIN pg_attribute a ON a.attrelid = d.adrelid AND a.attnum = d.adnum " + "WHERE d.adrelid = CAST(:t AS regclass) AND a.attname = 'processed_at'" + ), + {"t": table_name}, + ) + ).scalar_one() + processed_at_index = ( + await conn.execute( + text("SELECT indexdef FROM pg_indexes WHERE tablename = :t AND indexname = :i"), + {"t": table_name, "i": f"{table_name}_processed_at_idx"}, + ) + ).scalar_one() + finally: + async with pg_engine.begin() as conn: + await conn.run_sync(metadata.drop_all) + + assert primary_key == ["scope", "message_id"] + assert processed_at_default == "now()" + assert processed_at_index.endswith("(processed_at)") + + +def test_table_name_whose_index_name_exceeds_63_bytes_is_rejected() -> None: + with pytest.raises(ValueError, match="63 bytes"): + make_inbox_table(MetaData(), table_name="t" * 50)