Skip to content

fix(filesystem): consume messages in publish order - #2818

Open
arieleven wants to merge 1 commit into
celery:mainfrom
arieleven:fix/filesystem-fifo-order
Open

arieleven wants to merge 1 commit into
celery:mainfrom
arieleven:fix/filesystem-fifo-order

Conversation

@arieleven

Copy link
Copy Markdown

Requested by Taltal · project thread

Problem

The filesystem transport names each message file <ms>_<uuid4>.<queue>.msg, where <ms> is int(round(monotonic() * 1000)), and _get() takes the first name after sorted(os.listdir(...)). That is a text sort, so:

  • messages published within the same millisecond come back in uuid order, which is random. A producer that publishes 20 messages in a loop gets them back as, for example, [0, 5, 4, 2, 3, 1, 6, 12, 7, 10, ...];
  • the order also breaks when the millisecond counter gains a digit, because 10000_... sorts before 9999_....

Fix

Zero-pad the millisecond prefix and add a per-process sequence number (itertools.count()) between it and the uuid. Within one process the names now sort in publish order. Across processes, order is still by millisecond, as before. _get(), _size() and _purge() are unchanged: they still match on what follows the first ..

Upgrade note: while files with the old unpadded names are still in the folder, new files (with a leading 0) sort before them. Old messages are still delivered; only their order relative to new ones changes during the switch.

Tests

test_messages_are_consumed_in_publish_order patches monotonic for two cases: ten messages in the same millisecond, and the counter going from 9999 to 10004. Both fail on main (7dca15a6) and pass with the fix.

  • t/unit/transport/test_filesystem.py: 221 passed.
  • t/unit (with --continue-on-collection-errors, because the SQS/Azure extras are not installed here): 1573 passed, 186 skipped. One failure, test_FilesystemOpenErrors::test_get_closes_file_when_read_fails, fails on main too, and only in the full run (it passes when the file runs alone).
  • flake8 and pydocstyle are clean on the changed files. mypy kombu/transport/filesystem.py reports the same 22 errors as on main.

I used Claude Code to help find and write this fix. I reviewed and tested the change.

Message files were named <ms>_<uuid4>.<queue>.msg and _get() sorts the
names as text, so messages published in the same millisecond came out in
uuid (random) order, and the order also broke whenever the millisecond
counter gained a digit (9999 sorts after 10000).

Zero-pad the millisecond prefix and add a per-process sequence number
before the uuid.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant