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..b21145a8a2 100644 --- a/test/plugin/test_buffer.rb +++ b/test/plugin/test_buffer.rb @@ -435,6 +435,33 @@ def create_chunk_es(metadata, es) assert_equal({}, @p.dequeued) 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, m1.metadata + assert_equal [@dm1, @dm1], @p.queue.map(&:metadata) + assert_equal({m1.unique_id => m1}, @p.dequeued) + + 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. + # 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 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) + + 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)