From c739af6243ed4a967216a791fe341d8035ea1924 Mon Sep 17 00:00:00 2001 From: Mehrdad Biukian Naeini Date: Fri, 21 Aug 2026 21:18:46 +0400 Subject: [PATCH 1/3] fix(buffer): decrement queue_size in ensure so failed purge does not leak (#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 Signed-off-by: Mehrdad Biukian Naeini --- lib/fluent/plugin/buffer.rb | 10 ++++++++-- test/plugin/test_buffer.rb | 26 ++++++++++++++++++++++++++ 2 files changed, 34 insertions(+), 2 deletions(-) diff --git a/lib/fluent/plugin/buffer.rb b/lib/fluent/plugin/buffer.rb index ea50eb3ecb..243065be38 100644 --- a/lib/fluent/plugin/buffer.rb +++ b/lib/fluent/plugin/buffer.rb @@ -601,13 +601,19 @@ def purge_chunk(chunk_id) metadata = chunk.metadata log.on_trace { log.trace "purging a chunk", instance: self.object_id, chunk_id: dump_unique_id_hex(chunk_id), metadata: metadata } + bytesize = chunk.bytesize begin - bytesize = chunk.bytesize chunk.purge - @queue_size_metrics.sub(bytesize) rescue => e log.error "failed to purge buffer chunk", chunk_id: dump_unique_id_hex(chunk_id), error_class: e.class, error: e log.error_backtrace + ensure + # Always release the queued byte counter, even when purge raises + # (e.g. unlink/close failure on the buffer path). Otherwise + # @queue_size is never decremented for a dequeued chunk that is + # gone from both @queue and @dequeued, so the leak ratchets toward + # total_limit_size and storable? becomes permanently false (#5468). + @queue_size_metrics.sub(bytesize) end @dequeued_num[chunk.metadata] -= 1 diff --git a/test/plugin/test_buffer.rb b/test/plugin/test_buffer.rb index 9cb803f2f7..1318cdd4a6 100644 --- a/test/plugin/test_buffer.rb +++ b/test/plugin/test_buffer.rb @@ -435,6 +435,32 @@ def create_chunk_es(metadata, es) assert_equal({}, @p.dequeued) 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) + + queued_before = @p.queue_size + assert queued_before > 0 + + # 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? + end + test '#takeback_chunk returns false if specified chunk_id is already purged' do assert_equal [@dm0,@dm1,@dm1], @p.queue.map(&:metadata) assert_equal({}, @p.dequeued) From d6933e56b952f177be4e01497e1fabaf57696c6f Mon Sep 17 00:00:00 2001 From: Mehrdad Biukian Naeini Date: Wed, 26 Aug 2026 14:54:17 +0400 Subject: [PATCH 2/3] test: fix purge_chunk regression test per maintainer review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- test/plugin/test_buffer.rb | 16 ++++++---------- 1 file changed, 6 insertions(+), 10 deletions(-) diff --git a/test/plugin/test_buffer.rb b/test/plugin/test_buffer.rb index 1318cdd4a6..55b831660a 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? From 3d5b77e6c16662e6f5a20d59e561282ac942a1c3 Mon Sep 17 00:00:00 2001 From: Mehrdad Biukian Naeini Date: Sun, 30 Aug 2026 13:33:59 +0400 Subject: [PATCH 3/3] test: fix purge_chunk regression test per maintainer review (#5468) Correct the post-dequeue queue assertion (2 elements, not 3) and verify queue_size is released when chunk.purge raises during purge_chunk. Drops the chunk-flag assertion which is outside the scope of the #5468 leak fix. Signed-off-by: Mehrdad Biukian Naeini --- test/plugin/test_buffer.rb | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/test/plugin/test_buffer.rb b/test/plugin/test_buffer.rb index 55b831660a..b21145a8a2 100644 --- a/test/plugin/test_buffer.rb +++ b/test/plugin/test_buffer.rb @@ -436,8 +436,10 @@ def create_chunk_es(metadata, es) end test '#purge_chunk releases queue_size even when chunk.purge raises (#5468)' do + # Initial queue: [@dm0, @dm1, @dm1] m1 = @p.dequeue_chunk - assert_equal [@dm0, @dm1, @dm1], @p.queue.map(&:metadata) + assert_equal @dm0, m1.metadata + assert_equal [@dm1, @dm1], @p.queue.map(&:metadata) assert_equal({m1.unique_id => m1}, @p.dequeued) queued_before = @p.queue_size @@ -445,14 +447,17 @@ def create_chunk_es(metadata, es) # Simulate a physical purge failure (unlink/close on the buffer path) # so chunk.purge raises after the chunk is dequeued. + # Inject the failure on the single chunk instance under test (the file's + # established idiom) rather than mutating a shared class. (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 + # purge_chunk swallows the error but must still decrement queue_size once + # (the chunk is gone from both @queue and @dequeued), so the counter does + # not leak upward and storable? does not permanently flip to false (#5468). @p.purge_chunk(m1.unique_id) - # queue_size must return to its pre-dequeue value, not leak upward assert_equal queued_before - m1.bytesize, @p.queue_size assert @p.storable? end