From c48be0f2ac5be254501486188b0299e0fbdb51a4 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Wed, 28 Oct 2026 21:46:43 +0300 Subject: [PATCH 1/2] feat: add transactional inbox ledger Inbox(session, table) records processed messages by (scope, message_id) in the caller's transaction, with database-clock cleanup in bounded batches. Tests run against PostgreSQL 17 and cover concurrent claims, rollback, cancellation, commit failure and cleanup. --- .dockerignore | 14 +++ .github/workflows/_checks.yml | 124 +++++++++++++++++++ .github/workflows/ci.yml | 15 +++ .github/workflows/release.yml | 55 +++++++++ .github/workflows/scheduled.yml | 20 ++++ .gitignore | 22 ++++ AGENTS.md | 39 ++++++ CLAUDE.md | 1 + CONTEXT.md | 38 ++++++ Dockerfile | 24 ++++ LICENSE | 21 ++++ README.md | 182 ++++++++++++++++++++++++++++ docker-compose.yml | 29 +++++ justfile | 42 +++++++ pyproject.toml | 99 +++++++++++++++ sqlalchemy_inbox/__init__.py | 5 + sqlalchemy_inbox/inbox.py | 86 ++++++++++++++ sqlalchemy_inbox/py.typed | 0 sqlalchemy_inbox/schema.py | 41 +++++++ tests/__init__.py | 0 tests/conftest.py | 77 ++++++++++++ tests/test_claim.py | 126 ++++++++++++++++++++ tests/test_cleanup.py | 204 +++++++++++++++++++++++++++++++ tests/test_concurrency.py | 205 ++++++++++++++++++++++++++++++++ tests/test_schema.py | 56 +++++++++ 25 files changed, 1525 insertions(+) create mode 100644 .dockerignore create mode 100644 .github/workflows/_checks.yml create mode 100644 .github/workflows/ci.yml create mode 100644 .github/workflows/release.yml create mode 100644 .github/workflows/scheduled.yml create mode 100644 .gitignore create mode 100644 AGENTS.md create mode 100644 CLAUDE.md create mode 100644 CONTEXT.md create mode 100644 Dockerfile create mode 100644 LICENSE create mode 100644 README.md create mode 100644 docker-compose.yml create mode 100644 justfile create mode 100644 pyproject.toml create mode 100644 sqlalchemy_inbox/__init__.py create mode 100644 sqlalchemy_inbox/inbox.py create mode 100644 sqlalchemy_inbox/py.typed create mode 100644 sqlalchemy_inbox/schema.py create mode 100644 tests/__init__.py create mode 100644 tests/conftest.py create mode 100644 tests/test_claim.py create mode 100644 tests/test_cleanup.py create mode 100644 tests/test_concurrency.py create mode 100644 tests/test_schema.py 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..8848f40 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,99 @@ +[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. + "sqlalchemy[asyncio]>=2.0; 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) From 1193be7e580e41abc0ea40d06c973bec52ef2ca6 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Wed, 28 Oct 2026 23:07:28 +0300 Subject: [PATCH 2/2] fix: raise SQLAlchemy floor below Python 3.13 to 2.0.20 db-retry 0.9, a dev dependency, requires sqlalchemy>=2.0.20, so the floors job could not install 2.0.0 and that floor was never tested. --- pyproject.toml | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 8848f40..941c169 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -24,7 +24,8 @@ classifiers = [ 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. - "sqlalchemy[asyncio]>=2.0; python_version < '3.13'", + # 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.