fix(buffer): decrement queue_size in ensure so failed purge does not leak (#5468) - #5475
fix(buffer): decrement queue_size in ensure so failed purge does not leak (#5468)#5475mehrdadbn9 wants to merge 2 commits into
Conversation
|
@mehrdadbn9 Please write the PR description according to the PR template. |
3088b61 to
3300408
Compare
There was a problem hiding this comment.
Pull request overview
Fixes Fluentd buffer accounting so @queue_size_metrics is decremented even when chunk.purge raises, preventing storable? from becoming permanently false due to a leaked queued-byte counter (issue #5468).
Changes:
- Move
bytesize = chunk.bytesizeoutside thebegin/rescueand decrement@queue_size_metricsin anensureblock inBuffer#purge_chunk. - Add a regression test intended to verify
queue_sizeis released even whenchunk.purgeraises.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.
| File | Description |
|---|---|
| lib/fluent/plugin/buffer.rb | Ensures queued-byte accounting is always released during purge_chunk, even on purge errors. |
| test/plugin/test_buffer.rb | Adds a regression test for queue-size leak prevention when chunk.purge raises. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| # Simulate a physical purge failure (unlink/close on the buffer path) | ||
| # so chunk.purge raises after the chunk is dequeued. | ||
| failing_purge = Module.new do | ||
| def purge | ||
| raise IOError, 'simulated purge failure (unlink EIO)' | ||
| end | ||
| end | ||
| @p.buffer_class::Chunk.include(failing_purge) | ||
|
|
||
| m1 = @p.dequeue_chunk |
Watson1978
left a comment
Watson1978
left a comment
There was a problem hiding this comment.
When fix test codes, I think we can merge this PR.
| raise IOError, 'simulated purge failure (unlink EIO)' | ||
| end | ||
| end | ||
| @p.buffer_class::Chunk.include(failing_purge) |
There was a problem hiding this comment.
The new regression test calls @p.buffer_class::Chunk, but buffer_class is not defined anywhere in the codebase (neither on Fluent::Plugin::Buffer nor the test's DummyPlugin), so the test errors before any assertion runs.
| @p.buffer_class::Chunk.include(failing_purge) | ||
|
|
||
| m1 = @p.dequeue_chunk | ||
| assert_equal [@dm0, @dm1, @dm1], @p.queue.map(&:metadata) |
There was a problem hiding this comment.
Even after fixing the buffer_class typo, the test is still broken: the queue-state assertion right after one dequeue_chunk call expects the pre-dequeue 3-element array, and — depending how buffer_class::Chunk is resolved — either the simulated purge failure never actually triggers, or the later purged/queue_size assertions fail for structural reasons.
| # Simulate a physical purge failure (unlink/close on the buffer path) | ||
| # so chunk.purge raises after the chunk is dequeued. | ||
| failing_purge = Module.new do | ||
| def purge | ||
| raise IOError, 'simulated purge failure (unlink EIO)' | ||
| end | ||
| end |
There was a problem hiding this comment.
The test's failure-injection technique mutates a shared class rather than the single chunk instance under test, unlike the file's established idiom for the same purpose elsewhere.
suggestion:
diff --git a/test/plugin/test_buffer.rb b/test/plugin/test_buffer.rb
index 1318cdd4..55b83166 100644
--- a/test/plugin/test_buffer.rb
+++ b/test/plugin/test_buffer.rb
@@ -436,15 +436,6 @@ def create_chunk_es(metadata, es)
end
test '#purge_chunk releases queue_size even when chunk.purge raises (#5468)' do
- # Simulate a physical purge failure (unlink/close on the buffer path)
- # so chunk.purge raises after the chunk is dequeued.
- failing_purge = Module.new do
- def purge
- raise IOError, 'simulated purge failure (unlink EIO)'
- end
- end
- @p.buffer_class::Chunk.include(failing_purge)
-
m1 = @p.dequeue_chunk
assert_equal [@dm0, @dm1, @dm1], @p.queue.map(&:metadata)
assert_equal({m1.unique_id => m1}, @p.dequeued)
@@ -452,10 +443,15 @@ def purge
queued_before = @p.queue_size
assert queued_before > 0
+ # Simulate a physical purge failure (unlink/close on the buffer path)
+ # so chunk.purge raises after the chunk is dequeued.
+ (class << m1; self; end).module_eval do
+ define_method(:purge) { raise IOError, 'simulated purge failure (unlink EIO)' }
+ end
+
# purge_chunk swallows the error but must still decrement queue_size
@p.purge_chunk(m1.unique_id)
- assert m1.purged
# queue_size must return to its pre-dequeue value, not leak upward
assert_equal queued_before - m1.bytesize, @p.queue_size
assert @p.storable?
…leak (fluent#5468) When chunk.purge raises (e.g. unlink/close failure on the buffer path), the rescue swallows the error but @queue_size_metrics.sub was skipped. The chunk is already gone from @Dequeued, so it is never retried and the queued byte counter ratchets toward total_limit_size, making storable? permanently false and causing spurious BufferOverflowError on a near-empty buffer. Moving the sub into an ensure block keeps the counter correct regardless of purge success. Co-Authored-By: Mehrdad Biukian <mehrdad@example.com> Signed-off-by: Mehrdad Biukian Naeini <mehrdadbiukian@gmail.com>
- Use singleton-class monkey-patch on the specific chunk instance (established idiom in test_buffer.rb) instead of mutating a shared class. - Remove redundant assertion 'assert m1.purged' — purge_chunk swallows the error so the chunk is never marked purged; queue_size decrement is the critical invariant. - Keep queue_size assertion using bytesize captured before purge, matching the implementation in buffer.rb:595-620. Signed-off-by: Mehrdad Biukian Naeini <mehrdadbiukian@gmail.com>
9eb01b0 to
d6933e5
Compare
Which issue(s) this PR fixes:
Fixes #5468
What this PR does / why we need it:
In
Fluent::Plugin::Buffer#purge_chunk,@queue_size_metricsis only decremented insidebegin/rescue. Whenchunk.purgeraises (e.g. unlink/close failure on the buffer path), the error is logged and swallowed but thesubis skipped. The chunk is already removed from@dequeued, so it is never retried — the queued byte counter ratchets towardtotal_limit_size, makingstorable?permanentlyfalseand causing a spuriousBufferOverflowErroron a near-empty buffer.This moves
bytesize = chunk.bytesizeabove thebeginand moves@queue_size_metrics.sub(bytesize)into anensureblock so the counter is always released exactly once perpurge_chunkcall, regardless of whetherpurgesucceeds.Why safe:
ensureruns exactly once per call → no double-decrement.purge_chunkwas previously enqueued (dequeue/emit path), so thesubis always balanced.chunk.purgeraises, the chunk is lost on filesystem error (already a logged, separate failure); this change only preserves counter integrity sostorable?no longer spuriously goes false.Docs Changes:
None (behavior fix; inline comment added).
Release Note:
Fixed a bug where a failed chunk purge could leak
@queue_size_metricsand permanently disablestorable?(spurious BufferOverflowError).