Describe the bug
In Fluent::Plugin::Buffer#purge_chunk, the queued byte counter (@queue_size_metrics) is only decremented inside the begin/rescue block, after chunk.purge succeeds. If chunk.purge raises, the rescue swallows the error and @queue_size_metrics.sub(bytesize) is skipped — but the chunk has already been removed from @Dequeued at the top of the method, so nothing ever retries the purge. The result is a permanent leak of @queue_size (and, if the failure was in the file unlink, an orphaned buffer file left on disk).
Because storable? is
def storable?
@total_limit_size > @stage_size_metrics.get + @queue_size_metrics.get
end
each failed purge ratchets @queue_size upward and never comes back down. Over a long-lived process this drifts toward total_limit_size, at which point storable? returns false and every subsequent emit raises Fluent::Plugin::Buffer::BufferOverflowError even though the real staged + queued data is small or empty. Input plugins then start rejecting incoming data (e.g. in_tail stops advancing), so records are dropped. Only a restart resets the counter.
This is the queue-side sibling of the staged-side accounting leak fixed in #5439. Same user-visible symptom — spurious BufferOverflowError on a near-empty buffer — but a different trigger and code path.
def purge_chunk(chunk_id)
metadata = nil
synchronize do
chunk = @dequeued.delete(chunk_id) # (1) removed from @dequeued up front
return nil unless chunk
metadata = chunk.metadata
begin
bytesize = chunk.bytesize
chunk.purge # (2) can raise (unlink/close on the buffer path)
@queue_size_metrics.sub(bytesize) # (3) SKIPPED when (2) raises
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
end
@dequeued_num[chunk.metadata] -= 1
...
end
nil
end
If (2) raises:
(3) never runs → @queue_size is never decremented for those bytes.
The chunk is already gone from both @Queue (dequeued before the write) and @Dequeued (step 1) → no retry path, the leak is permanent.
If the raise came from the file unlink itself, the buffer file also stays on disk permanently.
To be precise about terminology: the leak is in the queued byte size (@queue_size / buffer_queue_byte_size), which is what storable? checks — not the queue element count.
If (2) raises:
(3) never runs → @queue_size is never decremented for those bytes.
The chunk is already gone from both @Queue (dequeued before the write) and @Dequeued (step 1) → no retry path, the leak is permanent.
If the raise came from the file unlink itself, the buffer file also stays on disk permanently.
To be precise about terminology: the leak is in the queued byte size (@queue_size / buffer_queue_byte_size), which is what storable? checks — not the queue element count.
To Reproduce
Self-contained script driving the real Buffer API (write → enqueue_chunk → dequeue_chunk → purge_chunk), with chunk.purge forced to raise to simulate an unlink/EIO failure on the buffer path. Deterministic — no flush-thread timing.
require 'fluent/version'
require 'fluent/test'
require 'fluent/test/driver/output'
require 'fluent/plugin/output'
require 'fluent/plugin/buffer'
require 'fluent/plugin/buffer/memory_chunk'
# simulate the physical purge (unlink/close) failing
module FailingPurge
def purge; raise IOError, "simulated purge failure (unlink EIO)"; end
end
Fluent::Plugin::Buffer::MemoryChunk.prepend(FailingPurge)
class PurgeLeakOut < Fluent::Plugin::Output
def format(tag, time, record); record['msg'] + "\n"; end
def write(chunk); end # write to the OUTPUT succeeds; only purge fails
end
Fluent::Plugin.register_output('purge_leak_out', PurgeLeakOut)
d = Fluent::Test::Driver::Output.new(PurgeLeakOut)
d.configure(%[
<buffer>
@type memory
chunk_limit_size 1k
total_limit_size 16k
flush_at_shutdown false
</buffer>
])
out = d.instance; out.start; buf = out.buffer
1.upto(40) do |i|
begin
meta = buf.metadata()
buf.write({ meta => ["x" * 900 + "\n"] }) # stage ~900B (succeeds)
buf.enqueue_chunk(meta) # staged -> queue (queue_size += ~900)
c = buf.dequeue_chunk # queue -> dequeued (flush picks it up)
buf.purge_chunk(c.unique_id) if c # commit -> purge; chunk.purge raises
note = ""
rescue Fluent::Plugin::Buffer::BufferOverflowError
note = "BufferOverflowError (buffer refuses write)"
end
real = buf.queue.map(&:bytesize).sum + buf.instance_variable_get(:@dequeued).values.map(&:bytesize).sum
printf("iter=%-3d queue_size=%-6d real_queued=%-3d storable?=%-6s %s\n", i, buf.queue_size, real, buf.storable?, note)
endRun it (no local ruby needed):
docker run --rm -v "$PWD:/t" ruby:3.2 sh -c \
'gem install fluentd -v 1.19.3 --no-document -q; ruby /t/repro.rb'
Observed output (fluentd 1.19.3)
total_limit_size: 16384 bytes
iter=1 queue_size=901 real_queued=0 storable?=true
iter=2 queue_size=1802 real_queued=0 storable?=true
...
iter=18 queue_size=16218 real_queued=0 storable?=true
iter=19 queue_size=17119 real_queued=0 storable?=false
iter=20 queue_size=17119 real_queued=0 storable?=false BufferOverflowError (buffer refuses write)
iter=21 queue_size=17119 real_queued=0 storable?=false BufferOverflowError (buffer refuses write)
... (every subsequent emit raises)
queue_size climbs by each failed chunk's bytesize and never comes back down, while real_queued (the bytes actually held in @queue + @dequeued) stays 0. Once queue_size crosses total_limit_size, storable? is permanently false and every emit raises BufferOverflowError on an empty buffer. Each iteration also logs failed to purge buffer chunk. A restart clears it.
Moving @queue_size_metrics.sub(bytesize) so it runs even when chunk.purge fails (i.e. in an ensure) makes queue_size stay at 0 and the overflow never occurs — verified in the same harness.
fluentd-purge-leak-repro.zip
Expected behavior
A failed chunk.purge should not permanently inflate @queue_size. The queued bytes should be reconciled (decremented, or the purge retried/deferred) so that total_queued_size stays consistent with the chunks the buffer is actually holding, and storable? does not become permanently false due to accounting drift.
Your Environment
- Fluentd version:reproduced by source inspection on master; the purge_chunk code path is identical in v1.18.0, v1.19.0, v1.19.3
- Package version:
- Operating system:
- Kernel version:
Your Configuration
<system>
log_level info
</system>
<source>
@type sample
tag test
sample {"msg":"hello world hello world hello world"}
rate 200
</source>
<match test>
@type noop_format
<buffer>
@type file
path /tmp/buf/test
chunk_limit_size 1k
total_limit_size 16k
flush_mode immediate
flush_thread_count 1
retry_type periodic
retry_wait 1s
overflow_action throw_exception
flush_at_shutdown false
</buffer>
</match>Your Error Log
# Trigger: chunk.purge raises; purge_chunk rescues it, so queue_size is never decremented
2026-08-10 08:43:50 +0000 [error]: #0 failed to purge buffer chunk chunk_id="658ad59b0aea892ad9316cb427274a28" error_class=IOError error=#<IOError: simulated purge failure (unlink EIO on buffer path)>
2026-08-10 08:43:50 +0000 [error]: #0 /path/fail_purge.rb:9:in `purge'
2026-08-10 08:43:50 +0000 [error]: #0 /usr/local/bundle/gems/fluentd-1.19.3/lib/fluent/plugin/buffer.rb:606:in `block in purge_chunk'
2026-08-10 08:43:50 +0000 [error]: #0 /usr/local/bundle/gems/fluentd-1.19.3/lib/fluent/plugin/buffer.rb:597:in `purge_chunk'
2026-08-10 08:43:50 +0000 [error]: #0 /usr/local/bundle/gems/fluentd-1.19.3/lib/fluent/plugin/output.rb:1139:in `commit_write'
2026-08-10 08:43:50 +0000 [error]: #0 /usr/local/bundle/gems/fluentd-1.19.3/lib/fluent/plugin/output.rb:1260:in `try_flush'
2026-08-10 08:43:50 +0000 [error]: #0 /usr/local/bundle/gems/fluentd-1.19.3/lib/fluent/plugin/output.rb:1571:in `flush_thread_run'
# Consequence: leaked queue_size crosses total_limit_size -> storable? false -> every emit rejected
2026-08-10 08:43:50 +0000 [warn]: #0 failed to write data into buffer by buffer overflow action=:throw_exception
2026-08-10 08:43:50 +0000 [warn]: #0 emit transaction failed: error_class=Fluent::Plugin::Buffer::BufferOverflowError error="buffer space has too many data" location="/usr/local/bundle/gems/fluentd-1.19.3/lib/fluent/plugin/buffer.rb:335:in `write'" tag="test"
2026-08-10 08:43:50 +0000 [warn]: #0 /usr/local/bundle/gems/fluentd-1.19.3/lib/fluent/plugin/output.rb:921:in `emit_buffered'
2026-08-10 08:43:50 +0000 [warn]: #0 /usr/local/bundle/gems/fluentd-1.19.3/lib/fluent/plugin/in_sample.rb:115:in `emit'
Additional context
No response
Describe the bug
In Fluent::Plugin::Buffer#purge_chunk, the queued byte counter (@queue_size_metrics) is only decremented inside the begin/rescue block, after chunk.purge succeeds. If chunk.purge raises, the rescue swallows the error and @queue_size_metrics.sub(bytesize) is skipped — but the chunk has already been removed from @Dequeued at the top of the method, so nothing ever retries the purge. The result is a permanent leak of @queue_size (and, if the failure was in the file unlink, an orphaned buffer file left on disk).
Because storable? is
def storable?
@total_limit_size > @stage_size_metrics.get + @queue_size_metrics.get
end
each failed purge ratchets @queue_size upward and never comes back down. Over a long-lived process this drifts toward total_limit_size, at which point storable? returns false and every subsequent emit raises Fluent::Plugin::Buffer::BufferOverflowError even though the real staged + queued data is small or empty. Input plugins then start rejecting incoming data (e.g. in_tail stops advancing), so records are dropped. Only a restart resets the counter.
This is the queue-side sibling of the staged-side accounting leak fixed in #5439. Same user-visible symptom — spurious BufferOverflowError on a near-empty buffer — but a different trigger and code path.
If (2) raises:
(3) never runs → @queue_size is never decremented for those bytes.
The chunk is already gone from both @Queue (dequeued before the write) and @Dequeued (step 1) → no retry path, the leak is permanent.
If the raise came from the file unlink itself, the buffer file also stays on disk permanently.
To be precise about terminology: the leak is in the queued byte size (@queue_size / buffer_queue_byte_size), which is what storable? checks — not the queue element count.
If (2) raises:
(3) never runs → @queue_size is never decremented for those bytes.
The chunk is already gone from both @Queue (dequeued before the write) and @Dequeued (step 1) → no retry path, the leak is permanent.
If the raise came from the file unlink itself, the buffer file also stays on disk permanently.
To be precise about terminology: the leak is in the queued byte size (@queue_size / buffer_queue_byte_size), which is what storable? checks — not the queue element count.
To Reproduce
Self-contained script driving the real Buffer API (
write → enqueue_chunk → dequeue_chunk → purge_chunk), withchunk.purgeforced to raise to simulate an unlink/EIOfailure on the buffer path. Deterministic — no flush-thread timing.Run it (no local ruby needed):
Observed output (fluentd 1.19.3)
queue_sizeclimbs by each failed chunk's bytesize and never comes back down, whilereal_queued(the bytes actually held in@queue+@dequeued) stays 0. Oncequeue_sizecrossestotal_limit_size,storable?is permanently false and every emit raisesBufferOverflowErroron an empty buffer. Each iteration also logsfailed to purge buffer chunk. A restart clears it.Moving
@queue_size_metrics.sub(bytesize)so it runs even whenchunk.purgefails (i.e. in anensure) makesqueue_sizestay at 0 and the overflow never occurs — verified in the same harness.fluentd-purge-leak-repro.zip
Expected behavior
A failed chunk.purge should not permanently inflate @queue_size. The queued bytes should be reconciled (decremented, or the purge retried/deferred) so that total_queued_size stays consistent with the chunks the buffer is actually holding, and storable? does not become permanently false due to accounting drift.
Your Environment
Your Configuration
<system> log_level info </system> <source> @type sample tag test sample {"msg":"hello world hello world hello world"} rate 200 </source> <match test> @type noop_format <buffer> @type file path /tmp/buf/test chunk_limit_size 1k total_limit_size 16k flush_mode immediate flush_thread_count 1 retry_type periodic retry_wait 1s overflow_action throw_exception flush_at_shutdown false </buffer> </match>Your Error Log
Additional context
No response