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).
| } | ||
|
|
||
| /// 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