[fix][client] Fix batch container buffer ownership on the failure-recovery path - #26455
Open
nodece wants to merge 1 commit into
Open
[fix][client] Fix batch container buffer ownership on the failure-recovery path#26455nodece wants to merge 1 commit into
nodece wants to merge 1 commit into
Conversation
…overy path When compression or encryption releases the batch buffer before a flush that then fails, the old recovery reused the released buffer (use-after-free) or orphaned the compressed payload (leak). Track batch-buffer ownership explicitly instead of nulling the buffer inside the build method: - BatchMessageContainerImpl: add a batchPayloadOwned flag; a build only starts while the container owns its buffer. Compression/encryption that releases it clears the flag, and resetPayloadAfterFailedPublishing() reallocates instead of reusing released memory. Releasing the compressed payload when encryption fails fixes the orphaned-payload leak. - resetPayloadAfterFailedPublishing(): skip reallocation when the container has no messages left, since key-based batching forwards the reset to sub-batches that already succeeded and cleared. - RawBatchMessageContainerImpl: symmetrically release the compressed payload and any partially built encrypted buffer when encryption fails, and release the serialized payloads in toByteBuf() on failure instead of orphaning them. - StrategicTwoPhaseCompactor: always clear the batch container on a failed flush, including when the failure is an Error (discard(Exception) cannot take an Error). - Tests: cover failure recovery with/without compression, encryption-failure buffer release, and the fail-fast guard against re-entering a build without reset. Assisted-by: Claude Code Assisted-by: Codex
Contributor
|
One additional case worth considering is partial success during multi-batch construction. For example, sub-batch A successfully creates an Repeated retries may retain additional command buffers and cause direct-memory growth or eventually OOM. With |
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.
Motivation
When compression or encryption releases the batch buffer before a flush that then fails,
resetPayloadAfterFailedPublishing()could not tell whether the container still owned the buffer: retrying reused the released buffer (use-after-free), and the compressed/encrypted payload that replaced it was left orphaned (leak).Modifications
batchPayloadOwnedflag instead of inferring it from the buffer reference. A build only starts while the container owns its buffer; compression/encryption that releases it clears the flag, andresetPayloadAfterFailedPublishing()reallocates instead of reusing released memory. The compressed payload is released when encryption fails and the container no longer owns it (fixes the orphaned-payload leak).resetPayloadAfterFailedPublishing()also skips reallocation when the container has no messages left, since key-based batching forwards the reset to sub-batches that already succeeded and cleared.toByteBuf()on failure instead of orphaning them.Error(discard(Exception)cannot take anError).