[ericdelorefice]

The SQLAlchemy transport's Channel._purge deletes the queue's messages and commits, and when that commit raises OperationalError it rolls back and still returns the count. The caller is told the messages were purged when they are all still in the queue. This makes _purge re-raise after the rollback, the same change #2793 made to _put; #2793 listed _purge as not in that PR.

Problem

kombu/transport/sqlalchemy/__init__.py, Channel._purge:

try:
    self.session.commit()
except OperationalError:
    self.session.rollback()
return count

Queue.purge() and channel.queue_purge() return that count, so a purge that was rolled back reports N messages removed and the messages are delivered later.

The trigger I reproduced is SQLite's database is locked: another connection holds a read transaction (for example a monitoring query, a backup or an admin shell) past the busy timeout (5 seconds by default) while the purge commits. If the other connection holds a write lock instead, the DELETE itself fails, and that error already propagates on main, so only the commit path is affected. Other OperationalErrors raised at commit on other databases take the same path; I have only reproduced the SQLite one.

Reproduction

Real kombu, SQLAlchemy 2.1.6 and SQLite, nothing mocked, default busy timeout. Three messages are published to work, then a separate process opens the database with sqlite3, runs BEGIN and a SELECT, holds it and commits, while Queue.purge() runs.

On main (8c540f6):

== lock held 6 s (past the 5 s busy timeout)
purge() returned 3 after 5.2s
after: message_count 3
still delivered: [{'job': 0}, {'job': 1}, {'job': 2}]
== control: lock held 1 s
purge() returned 3 after 1.1s
after: message_count 0

With this change:

== lock held 6 s (past the 5 s busy timeout)
purge() raised OperationalError: (sqlite3.OperationalError) database is locked after 5.2s
after: message_count 3
== control: lock held 1 s
purge() returned 3 after 1.1s
after: message_count 0

Change

_purge re-raises after the rollback (3 added lines: a comment and the raise). On a failed commit the rollback leaves the session usable for the next call. A purge that fails now raises instead of returning a count; nothing is deleted in either case.

One side effect: queue_delete calls _purge, so after_reply_message_received, which runs in the finally of Mailbox._collect and collect_replies, now raises on a commit lock instead of silently leaving the reply queue in place. When that happens, the replies collected in that call are lost to the caller. I left the Mailbox side alone in this PR.

Verification

Two new tests in t/unit/transport/test_sqlalchemy.py, using a real SQLite file and a 0.05 s busy timeout, like the #2793 tests:

• test_purge_raises_when_commit_fails: a second sqlite3 connection holds a read transaction, so the DELETE runs but the COMMIT gets database is locked. Asserts the OperationalError and that all 3 messages are still in the queue.
• control test_purge_returns_count_when_commit_succeeds.

The first test fails on main and passes with the change (5 runs each). The file passes with SQLAlchemy 2.1.6 and with 1.4.54, the oldest version requirements/extras/sqlalchemy.txt allows (16 passed). The rest of t/unit is unchanged: 2335 passed with this change, 2333 on main; test_filesystem.py::test_FilesystemOpenErrors::test_get_closes_file_when_read_fails fails here on main too, as noted in #2793. pre-commit (flake8, isort, mypy, codespell) passes on the changed files.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 85.78%. Comparing base (ed1366d) to head (b49c2eb).

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2811      +/-   ##
==========================================
+ Coverage   85.76%   85.78%   +0.01%     
==========================================
  Files          80       80              
  Lines       11596    11597       +1     
  Branches     1403     1403              
==========================================
+ Hits         9945     9948       +3     
+ Misses       1430     1428       -2     
  Partials      221      221              

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.