fix(sdk): respect max_buffer_size when merging batches - #3933
Conversation
|
Thanks for the PR. It is labeled Slash commands (own line, regular comment) move it around the queue:
See CONTRIBUTING.md for details. |
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #3933 +/- ##
=============================================
- Coverage 83.87% 70.31% -13.56%
Complexity 1358 1358
=============================================
Files 1212 1212
Lines 166843 142867 -23976
Branches 134306 110454 -23852
=============================================
- Hits 139937 100459 -39478
- Misses 23266 38588 +15322
- Partials 3640 3820 +180
🚀 New features to boost your workflow:
|
| /// charged against the `max_buffer_size` budget until the merged batch has been written. | ||
| fn merge(&mut self, other: Self) { | ||
| self.inner.messages.extend(other.inner.messages); | ||
| self.bytes_permit.merge(other.bytes_permit); |
There was a problem hiding this comment.
OwnedSemaphorePermit::merge sums u32 permit counts, and after this change the sum is bounded only by max_buffer_size — which nothing validates.
IggyByteSize::from(u64) accepts any value and BackgroundConfig has no validator, so a max_buffer_size above 4 GiB is constructible. With batch_size(0) + batch_length(0) and a long linger, one shard buffer accumulates adjacent same-destination batches until the merged sum passes u32::MAX. Note this needs only normal-sized batches — each individual charge is lossless, e.g. 4096 x 1 MiB against a 6 GiB budget; it is the sum that wraps.
Reproduced on tokio 1.53.1:
- release: wraps silently and permanently destroys permits (6 GiB budget leaves 1.7 GiB available after one merge + drop), after which the default
BackpressureMode::Blockblockssend()forever with no error and no log. - debug: panics
attempt to add with overflowinside the shard task, where it is swallowed — theJoinHandleat line 135 is never awaited, so no test in this file can observe it.
Worth separating from the pre-existing issues nearby: this is a new failure mode, since nothing summed permits before this change. On master the same configuration merely under-enforces the budget and the producer keeps running.
Suggested fix is one line at construction — reject max_buffer_size > u32::MAX in BackgroundConfig / ProducerDispatcher::new. That also closes the related truncation at producer_dispatcher.rs:135, where IggyByteSize::as_bytes_u32() charges size mod 2^32 (exactly 0 at whole multiples of 4 GiB, so such a batch bypasses the budget entirely).
There was a problem hiding this comment.
@numinnex thanks for the review! Thats actually an interesting one, since tokios Semaphore is usize, but the OwnedSemaphorePermit is u32 and merge operates and returns OwnedSemaphorePermits. So as you say, we acquire from a usize and then merge on a u32, which then can silently overflow.
Seems odd to me, I will also ask tokio whether this is intended.
There was a problem hiding this comment.
@numinnex okay, so literally 8h ago someone ran into the same issue:
tokio-rs/tokio#8395
It seems like they gonna change it to usize. I would wait for that, since it solves our problem?
| } | ||
|
|
||
| /// Takes over `other`'s messages together with its byte permit, so the buffered bytes stay | ||
| /// charged against the `max_buffer_size` budget until the merged batch has been written. |
There was a problem hiding this comment.
This doc states the new contract correctly, but the public documentation of max_buffer_size no longer matches it.
core/sdk/src/clients/producer_config.rs:108-109 still says:
Upper bound for the bytes held in memory across all shards.
After this change permits are held through core.send_internal until the write returns, so the budget covers buffered plus in-flight bytes, and it now couples to max_in_flight. That is the user-visible semantic change this PR makes, and it is currently undocumented.
The practical consequence is that Block-mode producers can block where they previously did not, most visibly on the linger-only shape (batch_length(0) + batch_size(0)) that this repo's own integration tests use at core/integration/tests/sdk/producer/background.rs:135-136,177-178,216-217, which pins close to the whole budget for a full write RTT. Default config pins under 3% (32 MiB budget against a 1 MiB batch_size), so there is no out-of-the-box cliff.
Suggest amending the max_buffer_size doc to say the budget spans buffered and in-flight bytes until the write completes, and calling the behavior change out in the release notes.
2 participants
Which issue does this PR address?
Closes #3932
Rationale
Respect the
max_buffer_sizesuch that the BackpressureMode can kick in.What changed?
Merge both inner IggyMessages and the OwnedSemaphorePermits (bytes_permit), freeing only when written, not after merge.
Local Execution
AI Usage
None