[ericdelorefice]

The SQLAlchemy transport's Channel._put commits the new message row and, when that commit raises OperationalError, rolls back and returns normally. The caller is told the message was stored when it was not. This makes _put re-raise after the rollback.

Problem

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

self.session.add(message)
try:
    self.session.commit()
except OperationalError:
    self.session.rollback()

Everything that stores a message goes through _put. Each of these reports success for a message that is not in the database:

• Producer.publish() returns normally.
• Producer.publish(retry=True) has nothing to retry, although the transport lists OperationalError in connection_errors so that Connection.ensure can recover from it.
• message.requeue() returns normally and the message is acked, so it is gone.
• closing a channel with unacked messages restores them through _put; a failed restore is dropped silently instead of being logged as UNABLE TO RESTORE and written out by emergency_dump_state, which is what QoS.restore_unacked_once does when a restore raises.

The trigger I reproduced is SQLite's database is locked: another connection (another producer or worker process) holds the write lock for longer than the busy timeout (5 seconds by default) while the commit waits. The transport's own _get takes that lock with BEGIN IMMEDIATE on SQLite. 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 3.53.1, nothing mocked: a separate process opens the same database file with sqlite3, runs BEGIN IMMEDIATE, holds it for 6 seconds and commits, while kombu publishes one message to queue work. Same results on main (91ff754) and on kombu 5.6.2 from PyPI.

On main:

== default exchange (Celery default routing), no retry: another writer holds the lock past the busy timeout
  publish() returned normally after 5.2s
  queue "work" message_count afterwards: 0; get() -> None
== same, publish(retry=True)
  publish() returned normally after 5.2s
  queue "work" message_count afterwards: 0; get() -> None
== named direct exchange "jobs", no retry, same lock
  publish() returned normally after 5.2s
  queue "work" message_count afterwards: 0; get() -> None
== control: the lock is released inside the busy timeout
  publish() returned normally after 1.3s
  queue "work" message_count afterwards: 1; get() -> {'job': 1}

With this change:

== default exchange (Celery default routing), no retry: another writer holds the lock past the busy timeout
  publish() raised sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) database is locked after 5.2s
  queue "work" message_count afterwards: 0; get() -> None
== same, publish(retry=True)
  publish() returned normally after 6.0s
  queue "work" message_count afterwards: 1; get() -> {'job': 1}
== named direct exchange "jobs", no retry, same lock
  publish() raised sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) database is locked after 5.2s
  queue "work" message_count afterwards: 0; get() -> None
== control: the lock is released inside the busy timeout
  publish() returned normally after 1.3s
  queue "work" message_count afterwards: 1; get() -> {'job': 1}

message.requeue() under the same lock, then the lock released and the channel closed:

main:         requeue() returned normally; unacked 0; queue holds 0
this change:  requeue() raised OperationalError: (sqlite3.OperationalError) database is locked; unacked 1; queue holds 1

Change

_put re-raises after the rollback (3 added lines, one of them the raise). The rollback stays, so the session is usable for the next call.

Behaviour changes, intended:

• publish(), requeue() and the restore on close now raise (or, on close, log and dump) where they used to report success.
• A retried publish is at least once: if a commit raised after the database had in fact stored the row, or a topic exchange routes to several queues and a later one fails after earlier ones were stored, the retry can store a second copy. Before this change the failed copy was lost instead.
• On this transport Connection.ensure reconnects and retries once; a lock that is still held through a second busy timeout raises.

Not in this PR: _get_or_create and _purge have the same except OperationalError: rollback() around their commits. Separately, with retry=True on a producer whose own named exchange is auto-declared, the retry now fails with RecoverableConnectionError('channel disconnected') instead of reaching _put, because the exchange in the retried declare list is still bound to the closed channel (_imaybe_declare in kombu/common.py raises there). I can look at that one separately if it is useful.

Verification

New tests in t/unit/transport/test_sqlalchemy.py. Each uses a real SQLite file, a second sqlite3 connection holding the write lock and a 0.05 s busy timeout.

• test_put_raises_when_commit_fails, test_publish_raises_when_put_fails
• test_publish_retry_stores_message_after_failed_put (the retry policy's errback releases the lock, so the retry stores the message)
• test_requeue_keeps_message_unacked_when_put_fails (the message is restored when the channel closes)
• test_close_dumps_message_when_restore_fails (emergency_dump_state gets the message)
• control test_put_after_failed_commit_succeeds: after the lock is released, the next _put is stored and is the only message in the queue

On main the first five fail and the control passes (5 failed, 9 passed in the file); with the change the file passes (14 passed), also with SQLAlchemy 1.4.54, the oldest version requirements/extras/sqlalchemy.txt allows. The rest of t/unit is unchanged (1563 passed with this change; test_filesystem.py::test_FilesystemOpenErrors::test_get_closes_file_when_read_fails fails here on main too). pre-commit (flake8, isort, mypy, codespell) and pydocstyle pass on the changed files.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 85.75%. Comparing base (9d9094f) to head (ee33526).

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2793      +/-   ##
==========================================
+ Coverage   85.71%   85.75%   +0.04%     
==========================================
  Files          80       80              
  Lines       11593    11594       +1     
  Branches     1402     1402              
==========================================
+ Hits         9937     9943       +6     
+ Misses       1434     1430       -4     
+ Partials      222      221       -1     

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