Repository navigation
Keep segments mapped while messages from them are delivered - #2324
Merged
Merged
Conversation
|
No issues found. |
Member
Author
|
@codex review |
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
|
Codex Review: Didn't find any major issues. Chef's kiss. Reviewed commit: ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
If Codex has suggestions, it will comment; otherwise it will react with 👍. Codex can also answer questions or update the PR. Try commenting "@codex address that feedback". |
Deliveries yield envelopes outside @msg_store_lock, and a message's body and headers are slices into its segment's mmap. While a delivery was suspended in a socket write, the segment could be unmapped and the delivery would read freed memory, crashing the broker: - stream retention (a publish rolling a segment, a max-length, max-length-bytes or max-age policy, a purge) dropping the segment a slow stream consumer or the HTTP stream reader was reading (#2181) - a queue segment deleted during the delivery, e.g. because the message was acked or the queue purged meanwhile - a queue, stream or MQTT session closed or deleted during a basic.get, an HTTP get or an MQTT send (consumers are waited for by Queue#close) MFile now has a lease count. Envelope carries the segment its message is read from, and MessageStore#shift_leased? (StreamMessageStore# read_leased? for the HTTP stream reader) shifts the message under the lock and yields it outside it with the segment leased, releasing it afterwards. Queue#get, Stream#get and MQTT Session#get_packet deliver through it. MFile#close of a leased file truncates it right away and only unmaps it when the last lease is released, without truncating then, as the file may have been opened again by that time (e.g. a vhost restart). Files that are never leased close as before. The HTTP StreamReader also skips to the next segment, with the correct offset, if its segment was dropped while reading, instead of raising KeyError, and the aliveness check reads the body inside the basic_get block. Fixes #2181 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01JKYANz6GAxtpchSUaMX9wj
carlhoerberg
force-pushed
the
ccr-325fbd60-tcppsq
branch
from
October 5, 2026 08:20
7d7c1a8 to
faf85b8
Compare
carlhoerberg
changed the title
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01JKYANz6GAxtpchSUaMX9wj
The only envelopes built without one were in Queue#reject, which now expires by segment position, and Queue#read, which now gets its envelope from the new MessageStore#envelope. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01JKYANz6GAxtpchSUaMX9wj
carlhoerberg
added a commit
that referenced
this pull request
Oct 15, 2026
### What does this pull request do? Follow-up to #2324. `Queue#expire_msg` copied every dead-lettered message out of its segment (`MessageStore#copy`, one allocation and memcpy of the whole record) before routing it, because the routing happens outside `@msg_store_lock` while a concurrent purge or queue/vhost delete can unmap the segment. With #2324's segment leases that copy isn't needed: the message is routed straight from the mmap with its segment leased. The lease can't be scoped to the `route` call. When the expiry is part of an outer dead-lettering chain (e.g. a dead-lettered message overflowing a `max-length` destination, whose dropped head is dead-lettered in turn), `DeadLetterer#route` only queues the publishes and the outer chain's drain runs them after `route` has returned. So the lease is released in the routed callback, which runs after the dead-lettered copies are published, after `delete_message`; and if `route` raises before queueing the callback, it's released there. Header mutation (`x-death` etc.) is unaffected: a `Table` read from a segment is read-only and copies its buffer on the first write. The retry queue from #2254 used `MessageStore#copy` the same way when moving a rejected message to the retry queue (`Queue#redeliver_message`). It now leases the segment too. Here the lease only has to cover the `route_to_retry_queue` call, because `DelayedRetryQueue#delay` writes the message to its own store before returning. This removes `MessageStore#copy` (and the `PriorityMessageStore` override) and its specs. Also adds the missing `require "./rough_time"` to `message_store.cr`, which uses `RoughTime` since #2341 and otherwise doesn't compile on its own (the "Compile changed *.cr files independently" check). ### How can it be tested? - `spec/queue_dead_lettering_spec.cr`, "keeps a dead-lettered message mapped until a nested chain publishes it": a rejected message is dead-lettered into a `max-length` 1 queue, whose dropped head is dead-lettered in turn into a third queue as part of the same chain. The head arrives intact, and the middle queue's segment is unmapped once the queue is deleted, so the lease isn't leaked. The spec fails with the release removed. - `spec/delayed_retry_queue_spec.cr`, "should release the primary queue's segment after routing a message to the retry queue": a message moved to the retry queue leaves the primary queue's segment unleased, so deleting the queue unmaps it. The spec fails with the release removed.
carlhoerberg
added a commit
that referenced
this pull request
Oct 15, 2026
### Why Messages read from the store are zero-copy views into mmap'd segments. A view stays valid only while `@msg_store_lock` is held, or while its segment is leased (`Envelope#lease`, from #2324). A lease makes a purge, delete or close defer the unmap until the lease is released. Leases are opt-in: deliveries take one through `shift_with_lease?`, and dead-lettering and the retry queue do since #2345, but `MessageStore#first?` and `#[]` return unleased views. Two callers used such views after releasing the lock: - `time_to_message_expiration` took the head message under the lock, then logged it and computed its expiry outside the lock. - Logging reads the message's body and headers from the segment. - A purge, delete or close on another thread can unmap the segment in between. Reading after the unmap segfaults when debug logging is enabled. - `reject(requeue: true)`, in `redeliver_message`, read the message under the lock and checked `has_expired?` outside it. That check only reads fields already copied out of the segment today, but it is the same hazard. `purge` had a related problem. It decided outside the lock whether it could take the fast `purge_all` path: no unacked messages, and purging at least `message_count`. It then reported that stale `message_count` as the number of purged messages. A publish between the check and the lock was purged without being counted. A `max_count`-limited purge could also remove more than it was asked to. This is one of the steps towards running the default execution context on several threads. ### What - `time_to_message_expiration` and `redeliver_message` work out what they need (the expiry, whether the message has expired) inside the lock, and only return plain values. A lease would also work, but these are a few field reads with no I/O (apart from the debug log line, when it's enabled), so holding the lock is cheaper than the two atomic updates a lease takes. - `purge` makes its fast-path decision and takes its count under the lock. - Specs in `spec/queue_spec.cr`: - The purge accounting spec fails reliably on `main`. - The unmap race spec only reproduces sometimes, because the window is the few microseconds the log line takes. - Still open, for a separate fix: `get`, `ack` and `reject` still update unacked counts outside the lock, so the `purge_all` check can still race a delivery or an ack on another thread. Since #2324 the delivery's lease keeps the purged segment mapped, so that race no longer reads unmapped memory. The purge can still delete the segment of a message that's being delivered, though, and the message's ack or requeue then refers to a deleted segment.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What does this pull request do?
Fixes #2181, and the same class of bug for queues and MQTT sessions. Includes what was #2325 and #2333.
Deliveries yield envelopes outside
@msg_store_lock, and a message's body and headers (the headersTablekeeps a pointer into its buffer) are slices into its segment's mmap. While a delivery was suspended in a socket write, the segment could be unmapped and the delivery would read freed memory and crash the broker:max-length/max-length-bytes/max-agepolicy or a purge runs retention, which unmapped the segment a slow stream consumer or the HTTPStreamReaderwas reading.basic.get, an HTTP get or an MQTT send. Consumers were already safe, asQueue#closewaits for their deliver loops.The fix is a lease count on
MFile, so a mapping stays alive while a delivery uses it:Envelopeholds (privately, and required) the segment its message is read from. The two places that built envelopes without one now don't:Queue#rejectexpires by segment position, andQueue#read(basic.recoverwithout requeue) uses the newMessageStore#envelope.MessageStore#shift_with_lease?(lock, consumer = nil) { |env| }shifts the message under the lock and yields it outside the lock with the segment leased, releasing it in anensure;StreamMessageStore#read_with_lease?does the same forStreamReader.Queue#get,Stream#getand MQTTSession#get_packetdeliver through them, so no delivery path handles the lease itself. Plainshift?stays for callers that use the envelope under the lock (purge, expiry, max-length), which don't need a lease.MFile#closeof a leased file truncates it to its size right away and only unmaps it when the last lease is released. That deferred unmap doesn't truncate, as the file may have been opened again by then (e.g. a vhost restart) and truncating would cut off the new store's writes. Files that are never leased close exactly as before. Deleted segments are unlinked right away; their disk space is reclaimed at the unmap, so one in-flight delivery later.@segment_readers(#2250) is unchanged: it counts consumers positioned in a segment and decides when todontneedits pages, while the lease covers only an in-flight delivery and decides when it's safe tomunmap.Also:
StreamReaderskips to the next segment, with the correct offset, if its segment was dropped while reading, instead of raisingKeyError; the aliveness endpoint compares the body inside thebasic_getblock instead of after it.Performance
The per-message cost is two uncontended atomic operations, with no hash lookup, lock or allocation. Release builds, one consumer, prefetch 1000, timed in-process, builds alternated:
All within run-to-run noise. *Measured on an earlier version that also did a hash lookup per message, which has since been removed.
How can it be tested?
All of these fail on
main, except the one guarding the deferred close:spec/stream_queue_spec.cr: the repro from Stream delivery reads segment mmap outside @msg_store_lock (use-after-munmap SIGSEGV) #2181, with publishes rolling segments and with amax-length-bytespolicy; the segment is dropped, but still mapped and readable inside the delivery, and closed afterwards.spec/stream_reader_spec.cr: retention drops the segment mid-read; bodies are intact and offsets continuous.spec/queue_spec.cr: a message acked from within its own delivery (normal and priority queue); a no-ack delivery during which another delivery moves the read segment on and the queue is purged; a queue closed during abasic.get. A segment reopened by a new store before the deferred unmap isn't truncated by it (this one passes onmaintoo, but fails with a deferred close that truncates by path).spec/mqtt/integrations/publish_spec.cr: a subscriber that stops reading while a 32 MB payload is sent, then the session is closed; the payload still arrives intact.spec/mfile_spec.cr: closing a leased file truncates it right away and unmaps when the last lease is released, without truncating a file appended to meanwhile.🤖 Generated with Claude Code
https://claude.ai/code/session_01JKYANz6GAxtpchSUaMX9wj