From 5ee23321754ceaa4e2a5ffdab921e8f235dff9ef Mon Sep 17 00:00:00 2001 From: Alexander Enrique Urieles Nieto Date: Fri, 17 Jul 2026 00:11:53 +0200 Subject: [PATCH] buffer: fix stage_size leak when a staged chunk is unstaged In Buffer#write_step_by_step, when a single record cannot be appended to an existing staged chunk without exceeding chunk_limit_size, the staged chunk is removed from @stage, marked unstaged and later enqueued via enqueue_unstaged_chunk. Unlike enqueue_chunk, enqueue_unstaged_chunk only adds to @queue_size and never subtracts from @stage_size, so the chunk's already-counted bytes remain in @stage_size forever. Over a long-lived process this makes @stage_size (and thus fluentd_output_status_buffer_total_bytes = @stage_size + @queue_size) drift upward, traffic-proportionally. Once it reaches total_limit_size, storable? returns false and every emit raises Fluent::Plugin::Buffer::BufferOverflowError even though the buffer is actually near-empty; input sources then reject incoming data. Subtract the chunk's original (already-counted) bytesize from @stage_size at the point of unstaging, so total_queued_size stays consistent with the real staged + queued bytes. Add a regression test asserting the invariant stage_size == sum of staged chunk sizes. Signed-off-by: Alexander Enrique Urieles Nieto --- lib/fluent/plugin/buffer.rb | 7 ++++++- test/plugin/test_buffer.rb | 17 +++++++++++++++++ 2 files changed, 23 insertions(+), 1 deletion(-) diff --git a/lib/fluent/plugin/buffer.rb b/lib/fluent/plugin/buffer.rb index 9607d0e60e..ea50eb3ecb 100644 --- a/lib/fluent/plugin/buffer.rb +++ b/lib/fluent/plugin/buffer.rb @@ -850,7 +850,12 @@ def write_step_by_step(metadata, data, format, splits_count, &block) # As already processed content is kept after rollback, then unstaged chunk should be queued. # After that, re-process current split again. # New chunk should be allocated, to do it, modify @stage and so on. - synchronize { @stage.delete(modified_metadata) } + synchronize do + @stage.delete(modified_metadata) + # Subtract this chunk's already-counted staged bytes here; + # otherwise @stage_size leaks upward over the buffer's lifetime. + @stage_size_metrics.sub(original_bytesize) if chunk.staged? + end staged_chunk_used = false chunk.unstaged! break diff --git a/test/plugin/test_buffer.rb b/test/plugin/test_buffer.rb index 6c62c3a542..9cb803f2f7 100644 --- a/test/plugin/test_buffer.rb +++ b/test/plugin/test_buffer.rb @@ -1339,6 +1339,23 @@ def create_chunk_es(metadata, es) c3 = create_chunk(m, ["a" * 128] * 6 + ["a" * 64]) assert !@p.chunk_size_full?(c3) end + + test '#write does not leak stage_size when a staged chunk is unstaged in write_step_by_step' do + # dm2 is staged with 896 bytes (7*128); chunk_limit_size is 1024. + # First record (64B) is appended to the 896B staged chunk (-> 960B), then + # the second record (200B) would exceed chunk_limit_size (1024), so + # write_step_by_step deletes the chunk from @stage, marks it unstaged and + # enqueues it via enqueue_unstaged_chunk. stage_size must drop by exactly + # the chunk's already-counted bytes (its 896B before this write), not by + # its current bytesize. + assert_equal 128*7 + 128*5, @p.stage_size + + dm2 = @p.metadata(timekey: @dm2.timekey) + @p.write({dm2 => ["x" * 64, "x" * 200]}) + + # Invariant: stage_size equals the real total size of the staged chunks. + assert_equal @p.stage.values.map(&:bytesize).sum, @p.stage_size + end end sub_test_case 'with configuration includes chunk_limit_records' do