From a73d7d4c0d2692ae0fc7f626faad4914bd7d29ce Mon Sep 17 00:00:00 2001
From: "Chris (ChrisJr404)" <11917633+ChrisJr404@users.noreply.github.com>
Date: Mon, 17 Aug 2026 22:49:29 -0400
Subject: [PATCH 1/6] Add evacuate_limit_size to bound chunk evacuation

When an output stays unreachable past the retry limit, queued chunks get
evacuated to the backup dir and can fill up the disk. evacuate_limit_size
caps the total evacuated size: once the budget is used up the remaining
chunks are purged without evacuating, and setting it to 0 turns evacuation
off entirely. Default stays nil (no limit) so existing behavior doesn't
change.

Signed-off-by: Chris (ChrisJr404) <11917633+ChrisJr404@users.noreply.github.com>
---
 lib/fluent/plugin/buffer.rb | 22 ++++++++++++++++--
 test/plugin/test_buffer.rb  | 45 +++++++++++++++++++++++++++++++++++++
 2 files changed, 65 insertions(+), 2 deletions(-)

diff --git a/lib/fluent/plugin/buffer.rb b/lib/fluent/plugin/buffer.rb
index a694db2000..c2dfa94c89 100644
--- a/lib/fluent/plugin/buffer.rb
+++ b/lib/fluent/plugin/buffer.rb
@@ -72,6 +72,9 @@ class BufferChunkOverflowError < BufferError; end # A record size is larger than
       desc 'If true, chunks are thrown away when unrecoverable error happens'
       config_param :disable_chunk_backup, :bool, default: false
 
+      desc 'The total size limit for chunks evacuated on unrecoverable errors. Once evacuation reaches this size, remaining chunks are purged without being evacuated. Set 0 to disable evacuation. nil (default) means no limit.'
+      config_param :evacuate_limit_size, :size, default: nil
+
       Metadata = Struct.new(:timekey, :tag, :variables, :seq) do
         def initialize(timekey, tag, variables)
           super(timekey, tag, variables, 0)
@@ -643,11 +646,15 @@ def clear_queue!
         log.on_trace { log.trace "clearing queue", instance: self.object_id }
 
         synchronize do
+          evacuated_size = 0
           until @queue.empty?
             begin
               q = @queue.shift
-              evacuate_chunk(q)
-              log.trace("purging a chunk in queue"){ {id: dump_unique_id_hex(chunk.unique_id), bytesize: chunk.bytesize, size: chunk.size} }
+              if evacuate_chunk?(q.bytesize, evacuated_size)
+                evacuate_chunk(q)
+                evacuated_size += q.bytesize
+              end
+              log.trace("purging a chunk in queue"){ {id: dump_unique_id_hex(q.unique_id), bytesize: q.bytesize, size: q.size} }
               q.purge
             rescue => e
               log.error "unexpected error while clearing buffer queue", error_class: e.class, error: e
@@ -658,6 +665,17 @@ def clear_queue!
         end
       end
 
+      # Decide whether a chunk should be evacuated, honoring evacuate_limit_size.
+      # nil limit means no bound (evacuate everything), 0 disables evacuation, and
+      # a positive limit only evacuates a chunk if it fits within the remaining budget
+      # so the total evacuated size never exceeds the limit.
+      def evacuate_chunk?(chunk_bytesize, evacuated_size)
+        return true if @evacuate_limit_size.nil?
+        return false if @evacuate_limit_size <= 0
+
+        evacuated_size + chunk_bytesize <= @evacuate_limit_size
+      end
+
       def evacuate_chunk(chunk)
         # Overwrite this on demand.
         #
diff --git a/test/plugin/test_buffer.rb b/test/plugin/test_buffer.rb
index 241ea20ee6..bf8e502067 100644
--- a/test/plugin/test_buffer.rb
+++ b/test/plugin/test_buffer.rb
@@ -513,6 +513,51 @@ def create_chunk_es(metadata, es)
       assert{ qchunks.all?{ |c| c.purged } }
     end
 
+    test '#clear_queue! evacuates every queued chunk by default (no evacuate_limit_size)' do
+      evacuated = []
+      @p.define_singleton_method(:evacuate_chunk) { |chunk| evacuated << chunk.bytesize }
+      qchunks = @p.queue.dup
+
+      assert_nil @p.evacuate_limit_size
+
+      @p.clear_queue!
+
+      assert_equal [100, 100, 3], evacuated
+      assert_equal [], @p.queue
+      assert{ qchunks.all?{ |c| c.purged } }
+    end
+
+    test '#clear_queue! skips evacuation entirely when evacuate_limit_size is 0, but still purges' do
+      p = create_buffer({'evacuate_limit_size' => 0})
+      evacuated = []
+      p.define_singleton_method(:evacuate_chunk) { |chunk| evacuated << chunk.bytesize }
+      p.start
+      qchunks = p.queue.dup
+
+      p.clear_queue!
+
+      assert_equal [], evacuated
+      assert_equal [], p.queue
+      assert{ qchunks.all?{ |c| c.purged } }
+    end
+
+    test '#clear_queue! bounds the total evacuated size to evacuate_limit_size' do
+      p = create_buffer({'evacuate_limit_size' => 200})
+      evacuated = []
+      p.define_singleton_method(:evacuate_chunk) { |chunk| evacuated << chunk.bytesize }
+      p.start
+      qchunks = p.queue.dup
+
+      p.clear_queue!
+
+      # queued chunks are 100, 100 and 3 bytes; the first two fill the 200 byte
+      # budget, so the last one is purged without being evacuated
+      assert_equal [100, 100], evacuated
+      assert_equal 200, evacuated.sum
+      assert_equal [], p.queue
+      assert{ qchunks.all?{ |c| c.purged } }
+    end
+
     test '#write returns immediately if argument data is empty array' do
       assert_equal [@dm0,@dm1,@dm1], @p.queue.map(&:metadata)
       assert_equal [@dm2,@dm3], @p.stage.keys

From aff7e2d0f88de0c6781ee244988ffd04e90fad30 Mon Sep 17 00:00:00 2001
From: Shizuo Fujita <fujita@clear-code.com>
Date: Mon, 19 Oct 2026 14:07:55 +0900
Subject: [PATCH 2/6] buffer: count already evacuated files toward
 evacuate_limit_size

The budget was reset on every clear_queue! call. A long outage hits
the retry limit many times, so the evacuated files kept growing past
the limit.

Now the budget is the limit minus the size of the files already in the
evacuation directory. Moving the files back for recovery frees the
budget again. Evacuation also stops at the first chunk that does not
fit, and a warning is logged for each chunk purged without being
evacuated.

Signed-off-by: Shizuo Fujita <fujita@clear-code.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 lib/fluent/plugin/buf_file.rb        |  2 +-
 lib/fluent/plugin/buf_file_single.rb |  2 +-
 lib/fluent/plugin/buffer.rb          | 42 ++++++++++++++++++----------
 test/plugin/test_buf_file.rb         | 34 ++++++++++++++++++++++
 test/plugin/test_buf_file_single.rb  | 34 ++++++++++++++++++++++
 test/plugin/test_buffer.rb           | 32 +++++++++++++++++++++
 6 files changed, 129 insertions(+), 17 deletions(-)

diff --git a/lib/fluent/plugin/buf_file.rb b/lib/fluent/plugin/buf_file.rb
index 42544273c3..8bf94c85ff 100644
--- a/lib/fluent/plugin/buf_file.rb
+++ b/lib/fluent/plugin/buf_file.rb
@@ -234,7 +234,7 @@ def evacuate_chunk(chunk)
           raise ArgumentError, "The chunk must be FileChunk, but it was #{chunk.class}."
         end
 
-        backup_dir = File.join(backup_base_dir, 'buffer', safe_owner_id)
+        backup_dir = evacuate_dir
         FileUtils.mkdir_p(backup_dir, mode: system_config.dir_permission || Fluent::DEFAULT_DIR_PERMISSION) unless Dir.exist?(backup_dir)
 
         FileUtils.copy([chunk.path, chunk.meta_path], backup_dir)
diff --git a/lib/fluent/plugin/buf_file_single.rb b/lib/fluent/plugin/buf_file_single.rb
index 3cd10bbf21..04216bbe5a 100644
--- a/lib/fluent/plugin/buf_file_single.rb
+++ b/lib/fluent/plugin/buf_file_single.rb
@@ -250,7 +250,7 @@ def evacuate_chunk(chunk)
           raise ArgumentError, "The chunk must be FileSingleChunk, but it was #{chunk.class}."
         end
 
-        backup_dir = File.join(backup_base_dir, 'buffer', safe_owner_id)
+        backup_dir = evacuate_dir
         FileUtils.mkdir_p(backup_dir, mode: system_config.dir_permission || Fluent::DEFAULT_DIR_PERMISSION) unless Dir.exist?(backup_dir)
 
         FileUtils.copy(chunk.path, backup_dir)
diff --git a/lib/fluent/plugin/buffer.rb b/lib/fluent/plugin/buffer.rb
index c2dfa94c89..e028212b3c 100644
--- a/lib/fluent/plugin/buffer.rb
+++ b/lib/fluent/plugin/buffer.rb
@@ -72,7 +72,7 @@ class BufferChunkOverflowError < BufferError; end # A record size is larger than
       desc 'If true, chunks are thrown away when unrecoverable error happens'
       config_param :disable_chunk_backup, :bool, default: false
 
-      desc 'The total size limit for chunks evacuated on unrecoverable errors. Once evacuation reaches this size, remaining chunks are purged without being evacuated. Set 0 to disable evacuation. nil (default) means no limit.'
+      desc 'The size limit of the evacuated chunk files in the evacuation directory. When the retry limit is reached, queued chunks are evacuated in order until this limit, and the rest are purged without being evacuated. Set 0 to disable evacuation. nil (default) means no limit.'
       config_param :evacuate_limit_size, :size, default: nil
 
       Metadata = Struct.new(:timekey, :tag, :variables, :seq) do
@@ -645,14 +645,26 @@ def purge_chunk(chunk_id)
       def clear_queue!
         log.on_trace { log.trace "clearing queue", instance: self.object_id }
 
+        if @evacuate_limit_size.nil?
+          evacuate_budget = nil
+        elsif @evacuate_limit_size > 0
+          evacuate_budget = @evacuate_limit_size - evacuated_bytesize
+        else
+          evacuate_budget = 0
+        end
+
         synchronize do
-          evacuated_size = 0
           until @queue.empty?
             begin
               q = @queue.shift
-              if evacuate_chunk?(q.bytesize, evacuated_size)
+              if evacuate_budget.nil?
                 evacuate_chunk(q)
-                evacuated_size += q.bytesize
+              elsif evacuate_budget > 0 && q.bytesize <= evacuate_budget
+                evacuate_chunk(q)
+                evacuate_budget -= q.bytesize
+              else
+                evacuate_budget = 0
+                log.warn "evacuate_limit_size is reached. chunk is purged without being evacuated.", chunk_id: dump_unique_id_hex(q.unique_id), evacuate_limit_size: @evacuate_limit_size
               end
               log.trace("purging a chunk in queue"){ {id: dump_unique_id_hex(q.unique_id), bytesize: q.bytesize, size: q.size} }
               q.purge
@@ -665,17 +677,6 @@ def clear_queue!
         end
       end
 
-      # Decide whether a chunk should be evacuated, honoring evacuate_limit_size.
-      # nil limit means no bound (evacuate everything), 0 disables evacuation, and
-      # a positive limit only evacuates a chunk if it fits within the remaining budget
-      # so the total evacuated size never exceeds the limit.
-      def evacuate_chunk?(chunk_bytesize, evacuated_size)
-        return true if @evacuate_limit_size.nil?
-        return false if @evacuate_limit_size <= 0
-
-        evacuated_size + chunk_bytesize <= @evacuate_limit_size
-      end
-
       def evacuate_chunk(chunk)
         # Overwrite this on demand.
         #
@@ -695,6 +696,13 @@ def evacuate_chunk(chunk)
         #       This feature enables it.
       end
 
+      def evacuated_bytesize
+        dir = evacuate_dir
+        Dir.children(dir).sum { |name| File.size?(File.join(dir, name)).to_i }
+      rescue SystemCallError
+        0
+      end
+
       def chunk_size_over?(chunk)
         chunk.bytesize > @chunk_limit_size || (@chunk_limit_records && chunk.size > @chunk_limit_records)
       end
@@ -1022,6 +1030,10 @@ def safe_owner_id
       def backup_base_dir
         system_config.root_dir || DEFAULT_BACKUP_DIR
       end
+
+      def evacuate_dir
+        File.join(backup_base_dir, 'buffer', safe_owner_id)
+      end
     end
   end
 end
diff --git a/test/plugin/test_buf_file.rb b/test/plugin/test_buf_file.rb
index 477494aad7..aa47358f0c 100644
--- a/test/plugin/test_buf_file.rb
+++ b/test/plugin/test_buf_file.rb
@@ -1696,5 +1696,39 @@ def evacuate_dir(plugin_id)
 
       assert_equal(expected_records, actual_records)
     end
+
+    test 'evacuate_limit_size bounds the evacuated files over repeated retry limits' do
+      plugin_id = "test_output"
+      tag = "test.1"
+      buffer_conf = {
+        "path" => @buf_dir,
+        "flush_mode" => "interval",
+        "flush_interval" => "1s",
+        "retry_type" => "periodic",
+        "retry_max_times" => 1,
+        "retry_randomize" => false,
+        "evacuate_limit_size" => 400,
+      }
+
+      configure_output(plugin_id, "tag", buffer_conf) do |output|
+        start_plugin(output)
+
+        reach_retry_limit = lambda do
+          errors = output.num_errors
+          emit_events(output, tag, dummy_event_stream)
+          wait { output.num_errors == errors + 1 }
+          proceed_to_next_retry(output)
+          wait { output.num_errors == errors + 2 }
+          wait { Dir.empty?(@buf_dir) }
+          Dir.glob(File.join(evacuate_dir(plugin_id), "*.log")).size
+        end
+
+        assert_equal [1, 2, 2, 2], Array.new(4) { reach_retry_limit.call }
+        assert_equal 2, output.log.out.logs.count { |l| l.include?("evacuate_limit_size is reached") }
+
+        FileUtils.rm(Dir.glob(File.join(evacuate_dir(plugin_id), "*")))
+        assert_equal 1, reach_retry_limit.call
+      end
+    end
   end
 end
diff --git a/test/plugin/test_buf_file_single.rb b/test/plugin/test_buf_file_single.rb
index 111d468bd4..984623a69f 100644
--- a/test/plugin/test_buf_file_single.rb
+++ b/test/plugin/test_buf_file_single.rb
@@ -1219,5 +1219,39 @@ def evacuate_dir(plugin_id)
 
       assert_equal(expected_records, actual_records)
     end
+
+    test 'evacuate_limit_size bounds the evacuated files over repeated retry limits' do
+      plugin_id = "test_output"
+      tag = "test.1"
+      buffer_conf = {
+        "path" => @buf_dir,
+        "flush_mode" => "interval",
+        "flush_interval" => "1s",
+        "retry_type" => "periodic",
+        "retry_max_times" => 1,
+        "retry_randomize" => false,
+        "evacuate_limit_size" => 300,
+      }
+
+      configure_output(plugin_id, "tag", buffer_conf) do |output|
+        start_plugin(output)
+
+        reach_retry_limit = lambda do
+          errors = output.num_errors
+          emit_events(output, tag, dummy_event_stream)
+          wait { output.num_errors == errors + 1 }
+          proceed_to_next_retry(output)
+          wait { output.num_errors == errors + 2 }
+          wait { Dir.empty?(@buf_dir) }
+          Dir.children(evacuate_dir(plugin_id)).size
+        end
+
+        assert_equal [1, 2, 2, 2], Array.new(4) { reach_retry_limit.call }
+        assert_equal 2, output.log.out.logs.count { |l| l.include?("evacuate_limit_size is reached") }
+
+        FileUtils.rm(Dir.glob(File.join(evacuate_dir(plugin_id), "*")))
+        assert_equal 1, reach_retry_limit.call
+      end
+    end
   end
 end
diff --git a/test/plugin/test_buffer.rb b/test/plugin/test_buffer.rb
index bf8e502067..566cf7cb1a 100644
--- a/test/plugin/test_buffer.rb
+++ b/test/plugin/test_buffer.rb
@@ -558,6 +558,38 @@ def create_chunk_es(metadata, es)
       assert{ qchunks.all?{ |c| c.purged } }
     end
 
+    test '#clear_queue! stops evacuating at the first chunk which does not fit in evacuate_limit_size' do
+      p = create_buffer({'evacuate_limit_size' => 150})
+      evacuated = []
+      p.define_singleton_method(:evacuate_chunk) { |chunk| evacuated << chunk.bytesize }
+      p.start
+      qchunks = p.queue.dup
+
+      p.clear_queue!
+
+      assert_equal [100], evacuated
+      assert_equal [], p.queue
+      assert{ qchunks.all?{ |c| c.purged } }
+      skipped_ids = qchunks.drop(1).map { |c| p.dump_unique_id_hex(c.unique_id) }
+      skipped_logs = $log.out.logs.select { |l| l.include?("evacuate_limit_size is reached. chunk is purged without being evacuated.") }
+      assert_equal skipped_ids, skipped_logs.map { |l| l[/chunk_id="(\h+)"/, 1] }
+    end
+
+    test '#clear_queue! counts already evacuated files toward evacuate_limit_size' do
+      p = create_buffer({'evacuate_limit_size' => 200})
+      evacuated = []
+      p.define_singleton_method(:evacuate_chunk) { |chunk| evacuated << chunk.bytesize }
+      p.define_singleton_method(:evacuated_bytesize) { 100 }
+      p.start
+      qchunks = p.queue.dup
+
+      p.clear_queue!
+
+      assert_equal [100], evacuated
+      assert_equal [], p.queue
+      assert{ qchunks.all?{ |c| c.purged } }
+    end
+
     test '#write returns immediately if argument data is empty array' do
       assert_equal [@dm0,@dm1,@dm1], @p.queue.map(&:metadata)
       assert_equal [@dm2,@dm3], @p.stage.keys

From 7b8d32c85f0ba0ae762a8291ab884bb04a11ccd9 Mon Sep 17 00:00:00 2001
From: Shizuo Fujita <fujita@clear-code.com>
Date: Mon, 19 Oct 2026 14:29:18 +0900
Subject: [PATCH 3/6] buffer: keep evacuating the chunks that fit in
 evacuate_limit_size

- Skip only the chunks that do not fit, so more data is kept under
  the same limit.
- Log one warning per clear_queue! instead of one per chunk, and none
  when the limit is 0.
- Do not let an error while counting the evacuated files escape
  clear_queue!. It stopped the flush thread and left the queue.
- Reject a negative limit, and warn when @id is not configured or the
  limit is smaller than chunk_limit_size.

Signed-off-by: Shizuo Fujita <fujita@clear-code.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 lib/fluent/plugin/buffer.rb         | 60 ++++++++++++++++++-----------
 test/plugin/test_buf_file.rb        |  2 +-
 test/plugin/test_buf_file_single.rb |  2 +-
 test/plugin/test_buffer.rb          | 58 +++++++++++++++++++++++++---
 4 files changed, 93 insertions(+), 29 deletions(-)

diff --git a/lib/fluent/plugin/buffer.rb b/lib/fluent/plugin/buffer.rb
index e028212b3c..4cfbc1c6d6 100644
--- a/lib/fluent/plugin/buffer.rb
+++ b/lib/fluent/plugin/buffer.rb
@@ -72,7 +72,7 @@ class BufferChunkOverflowError < BufferError; end # A record size is larger than
       desc 'If true, chunks are thrown away when unrecoverable error happens'
       config_param :disable_chunk_backup, :bool, default: false
 
-      desc 'The size limit of the evacuated chunk files in the evacuation directory. When the retry limit is reached, queued chunks are evacuated in order until this limit, and the rest are purged without being evacuated. Set 0 to disable evacuation. nil (default) means no limit.'
+      desc 'The size limit of the evacuated chunk files in the evacuation directory. When the retry limit is reached, queued chunks which fit in the remaining size are evacuated, and the others are purged without being evacuated. Set 0 to disable evacuation. nil (default) means no limit.'
       config_param :evacuate_limit_size, :size, default: nil
 
       Metadata = Struct.new(:timekey, :tag, :variables, :seq) do
@@ -230,6 +230,17 @@ def configure(conf)
         unless @queue_limit_length.nil?
           @total_limit_size = @chunk_limit_size * @queue_limit_length
         end
+        if @evacuate_limit_size
+          raise Fluent::ConfigError, "evacuate_limit_size must be 0 or larger" if @evacuate_limit_size < 0
+          if @evacuate_limit_size > 0
+            unless owner.plugin_id_configured?
+              log.warn "evacuate_limit_size does not count the chunk files evacuated before restart unless @id is configured"
+            end
+            if @evacuate_limit_size < @chunk_limit_size
+              log.warn "evacuate_limit_size is smaller than chunk_limit_size, so full chunks are never evacuated", evacuate_limit_size: @evacuate_limit_size, chunk_limit_size: @chunk_limit_size
+            end
+          end
+        end
         @stage_length_metrics = metrics_create(namespace: "fluentd", subsystem: "buffer", name: "stage_length",
                                                help_text: 'Length of stage buffers', prefer_gauge: true)
         @stage_length_metrics.set(0)
@@ -645,26 +656,25 @@ def purge_chunk(chunk_id)
       def clear_queue!
         log.on_trace { log.trace "clearing queue", instance: self.object_id }
 
-        if @evacuate_limit_size.nil?
-          evacuate_budget = nil
-        elsif @evacuate_limit_size > 0
-          evacuate_budget = @evacuate_limit_size - evacuated_bytesize
-        else
-          evacuate_budget = 0
-        end
+        evacuate_budget = nil
+        skipped_num = 0
+        skipped_bytesize = 0
 
         synchronize do
           until @queue.empty?
             begin
               q = @queue.shift
-              if evacuate_budget.nil?
+              if @evacuate_limit_size.nil?
                 evacuate_chunk(q)
-              elsif evacuate_budget > 0 && q.bytesize <= evacuate_budget
-                evacuate_chunk(q)
-                evacuate_budget -= q.bytesize
-              else
-                evacuate_budget = 0
-                log.warn "evacuate_limit_size is reached. chunk is purged without being evacuated.", chunk_id: dump_unique_id_hex(q.unique_id), evacuate_limit_size: @evacuate_limit_size
+              elsif @evacuate_limit_size > 0
+                evacuate_budget ||= @evacuate_limit_size - evacuated_bytesize
+                if q.bytesize <= evacuate_budget
+                  evacuate_chunk(q)
+                  evacuate_budget -= q.bytesize
+                else
+                  skipped_num += 1
+                  skipped_bytesize += q.bytesize
+                end
               end
               log.trace("purging a chunk in queue"){ {id: dump_unique_id_hex(q.unique_id), bytesize: q.bytesize, size: q.size} }
               q.purge
@@ -675,6 +685,10 @@ def clear_queue!
           end
           @queue_size_metrics.set(0)
         end
+
+        if skipped_num > 0
+          log.warn "evacuate_limit_size is reached. some chunks are purged without being evacuated.", evacuate_limit_size: @evacuate_limit_size, chunks: skipped_num, bytesize: skipped_bytesize
+        end
       end
 
       def evacuate_chunk(chunk)
@@ -696,13 +710,6 @@ def evacuate_chunk(chunk)
         #       This feature enables it.
       end
 
-      def evacuated_bytesize
-        dir = evacuate_dir
-        Dir.children(dir).sum { |name| File.size?(File.join(dir, name)).to_i }
-      rescue SystemCallError
-        0
-      end
-
       def chunk_size_over?(chunk)
         chunk.bytesize > @chunk_limit_size || (@chunk_limit_records && chunk.size > @chunk_limit_records)
       end
@@ -1034,6 +1041,15 @@ def backup_base_dir
       def evacuate_dir
         File.join(backup_base_dir, 'buffer', safe_owner_id)
       end
+
+      def evacuated_bytesize
+        dir = evacuate_dir
+        return 0 unless Dir.exist?(dir)
+        Dir.children(dir).sum { |name| File.size?(File.join(dir, name)).to_i }
+      rescue => e
+        log.warn "failed to get the size of the evacuated chunk files", dir: dir, error: e
+        0
+      end
     end
   end
 end
diff --git a/test/plugin/test_buf_file.rb b/test/plugin/test_buf_file.rb
index aa47358f0c..3f192e6400 100644
--- a/test/plugin/test_buf_file.rb
+++ b/test/plugin/test_buf_file.rb
@@ -1719,7 +1719,7 @@ def evacuate_dir(plugin_id)
           wait { output.num_errors == errors + 1 }
           proceed_to_next_retry(output)
           wait { output.num_errors == errors + 2 }
-          wait { Dir.empty?(@buf_dir) }
+          wait { output.retry.nil? }
           Dir.glob(File.join(evacuate_dir(plugin_id), "*.log")).size
         end
 
diff --git a/test/plugin/test_buf_file_single.rb b/test/plugin/test_buf_file_single.rb
index 984623a69f..ece5fddb68 100644
--- a/test/plugin/test_buf_file_single.rb
+++ b/test/plugin/test_buf_file_single.rb
@@ -1242,7 +1242,7 @@ def evacuate_dir(plugin_id)
           wait { output.num_errors == errors + 1 }
           proceed_to_next_retry(output)
           wait { output.num_errors == errors + 2 }
-          wait { Dir.empty?(@buf_dir) }
+          wait { output.retry.nil? }
           Dir.children(evacuate_dir(plugin_id)).size
         end
 
diff --git a/test/plugin/test_buffer.rb b/test/plugin/test_buffer.rb
index 566cf7cb1a..8cb0792ad9 100644
--- a/test/plugin/test_buffer.rb
+++ b/test/plugin/test_buffer.rb
@@ -539,6 +539,7 @@ def create_chunk_es(metadata, es)
       assert_equal [], evacuated
       assert_equal [], p.queue
       assert{ qchunks.all?{ |c| c.purged } }
+      assert{ $log.out.logs.none? { |l| l.include?("evacuate_limit_size is reached") } }
     end
 
     test '#clear_queue! bounds the total evacuated size to evacuate_limit_size' do
@@ -558,7 +559,7 @@ def create_chunk_es(metadata, es)
       assert{ qchunks.all?{ |c| c.purged } }
     end
 
-    test '#clear_queue! stops evacuating at the first chunk which does not fit in evacuate_limit_size' do
+    test '#clear_queue! skips the chunks which do not fit in evacuate_limit_size' do
       p = create_buffer({'evacuate_limit_size' => 150})
       evacuated = []
       p.define_singleton_method(:evacuate_chunk) { |chunk| evacuated << chunk.bytesize }
@@ -567,12 +568,12 @@ def create_chunk_es(metadata, es)
 
       p.clear_queue!
 
-      assert_equal [100], evacuated
+      assert_equal [100, 3], evacuated
       assert_equal [], p.queue
       assert{ qchunks.all?{ |c| c.purged } }
-      skipped_ids = qchunks.drop(1).map { |c| p.dump_unique_id_hex(c.unique_id) }
-      skipped_logs = $log.out.logs.select { |l| l.include?("evacuate_limit_size is reached. chunk is purged without being evacuated.") }
-      assert_equal skipped_ids, skipped_logs.map { |l| l[/chunk_id="(\h+)"/, 1] }
+      skipped_logs = $log.out.logs.select { |l| l.include?("evacuate_limit_size is reached") }
+      assert_equal 1, skipped_logs.size
+      assert{ skipped_logs.first.include?("chunks=1 bytesize=100") }
     end
 
     test '#clear_queue! counts already evacuated files toward evacuate_limit_size' do
@@ -590,6 +591,53 @@ def create_chunk_es(metadata, es)
       assert{ qchunks.all?{ |c| c.purged } }
     end
 
+    test '#clear_queue! purges all chunks even if it fails to get the size of the evacuated files' do
+      p = create_buffer({'evacuate_limit_size' => 200})
+      evacuated = []
+      p.define_singleton_method(:evacuate_chunk) { |chunk| evacuated << chunk.bytesize }
+      p.define_singleton_method(:evacuate_dir) { raise Encoding::CompatibilityError, "incompatible character encodings" }
+      p.start
+      qchunks = p.queue.dup
+
+      p.clear_queue!
+
+      assert_equal [100, 100], evacuated
+      assert_equal [], p.queue
+      assert{ qchunks.all?{ |c| c.purged } }
+      assert{ $log.out.logs.any? { |l| l.include?("failed to get the size of the evacuated chunk files") } }
+    end
+
+    test 'raises ConfigError if evacuate_limit_size is negative' do
+      assert_raise(Fluent::ConfigError) do
+        create_buffer({'evacuate_limit_size' => -1})
+      end
+    end
+
+    data(
+      'without @id' => [{}, true],
+      'with @id' => [{'@id' => 'test_output'}, false],
+    )
+    test 'warns that evacuate_limit_size needs @id to count the files evacuated before restart' do |(owner_conf, warned)|
+      buffer_conf = config_element('buffer', '', {'evacuate_limit_size' => '1g'})
+      owner = FluentPluginBufferTest::DummyOutputPlugin.new
+      owner.configure(config_element('ROOT', '', owner_conf, [buffer_conf]))
+      p = FluentPluginBufferTest::DummyPlugin.new
+      p.owner = owner
+      p.configure(buffer_conf)
+
+      assert_equal warned, $log.out.logs.any? { |l| l.include?("unless @id is configured") }
+    end
+
+    data(
+      'smaller than chunk_limit_size' => ['1m', true],
+      'larger than chunk_limit_size' => ['1g', false],
+    )
+    test 'warns if evacuate_limit_size is smaller than chunk_limit_size' do |(limit, warned)|
+      create_buffer({'evacuate_limit_size' => limit, 'chunk_limit_size' => '8m'})
+
+      assert_equal warned, $log.out.logs.any? { |l| l.include?("smaller than chunk_limit_size") }
+    end
+
     test '#write returns immediately if argument data is empty array' do
       assert_equal [@dm0,@dm1,@dm1], @p.queue.map(&:metadata)
       assert_equal [@dm2,@dm3], @p.stage.keys

From bcf89a6d74a748caf203b5819b1dcf4221743090 Mon Sep 17 00:00:00 2001
From: Shizuo Fujita <fujita@clear-code.com>
Date: Tue, 20 Oct 2026 10:55:40 +0900
Subject: [PATCH 4/6] buffer: do not evacuate chunks when the evacuated files
 cannot be counted

- Read the file names in the encoding of the evacuation directory, so
  non-ASCII paths can be counted under LANG=C.
- Purge chunks without evacuation if the evacuated files cannot be
  counted, so evacuate_limit_size keeps bounding the disk usage.
- Ignore evacuate_limit_size for buffers which do not evacuate chunks,
  such as the memory buffer.
- Fix the warnings about @id and chunk_limit_size to match the actual
  behavior.

Signed-off-by: Shizuo Fujita <fujita@clear-code.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 lib/fluent/plugin/buffer.rb         | 24 ++++++++++-----
 test/plugin/test_buf_file.rb        |  3 +-
 test/plugin/test_buf_file_single.rb |  3 +-
 test/plugin/test_buffer.rb          | 47 +++++++++++++++++++----------
 4 files changed, 51 insertions(+), 26 deletions(-)

diff --git a/lib/fluent/plugin/buffer.rb b/lib/fluent/plugin/buffer.rb
index 4cfbc1c6d6..9bd980dff4 100644
--- a/lib/fluent/plugin/buffer.rb
+++ b/lib/fluent/plugin/buffer.rb
@@ -232,12 +232,15 @@ def configure(conf)
         end
         if @evacuate_limit_size
           raise Fluent::ConfigError, "evacuate_limit_size must be 0 or larger" if @evacuate_limit_size < 0
-          if @evacuate_limit_size > 0
+          if !persistent?
+            log.warn "evacuate_limit_size is ignored because this buffer does not evacuate chunks"
+            @evacuate_limit_size = nil
+          elsif @evacuate_limit_size > 0
             unless owner.plugin_id_configured?
-              log.warn "evacuate_limit_size does not count the chunk files evacuated before restart unless @id is configured"
+              log.warn "evacuate_limit_size should be used with @id, because the evacuation directory without @id can change when the configuration is changed"
             end
             if @evacuate_limit_size < @chunk_limit_size
-              log.warn "evacuate_limit_size is smaller than chunk_limit_size, so full chunks are never evacuated", evacuate_limit_size: @evacuate_limit_size, chunk_limit_size: @chunk_limit_size
+              log.warn "evacuate_limit_size is smaller than chunk_limit_size, so full chunks may not be evacuated", evacuate_limit_size: @evacuate_limit_size, chunk_limit_size: @chunk_limit_size
             end
           end
         end
@@ -667,7 +670,7 @@ def clear_queue!
               if @evacuate_limit_size.nil?
                 evacuate_chunk(q)
               elsif @evacuate_limit_size > 0
-                evacuate_budget ||= @evacuate_limit_size - evacuated_bytesize
+                evacuate_budget ||= remaining_evacuate_size
                 if q.bytesize <= evacuate_budget
                   evacuate_chunk(q)
                   evacuate_budget -= q.bytesize
@@ -687,7 +690,7 @@ def clear_queue!
         end
 
         if skipped_num > 0
-          log.warn "evacuate_limit_size is reached. some chunks are purged without being evacuated.", evacuate_limit_size: @evacuate_limit_size, chunks: skipped_num, bytesize: skipped_bytesize
+          log.warn "some chunks are purged without being evacuated to stay within evacuate_limit_size.", evacuate_limit_size: @evacuate_limit_size, chunks: skipped_num, bytesize: skipped_bytesize
         end
       end
 
@@ -1044,10 +1047,15 @@ def evacuate_dir
 
       def evacuated_bytesize
         dir = evacuate_dir
-        return 0 unless Dir.exist?(dir)
-        Dir.children(dir).sum { |name| File.size?(File.join(dir, name)).to_i }
+        Dir.children(dir, encoding: dir.encoding).sum { |name| File.size?(File.join(dir, name)).to_i }
+      rescue Errno::ENOENT
+        0
+      end
+
+      def remaining_evacuate_size
+        @evacuate_limit_size - evacuated_bytesize
       rescue => e
-        log.warn "failed to get the size of the evacuated chunk files", dir: dir, error: e
+        log.warn "failed to get the size of the evacuated chunk files. queued chunks are purged without being evacuated.", error: e
         0
       end
     end
diff --git a/test/plugin/test_buf_file.rb b/test/plugin/test_buf_file.rb
index 3f192e6400..77f0333d74 100644
--- a/test/plugin/test_buf_file.rb
+++ b/test/plugin/test_buf_file.rb
@@ -1717,6 +1717,7 @@ def evacuate_dir(plugin_id)
           errors = output.num_errors
           emit_events(output, tag, dummy_event_stream)
           wait { output.num_errors == errors + 1 }
+          wait { output.retry }
           proceed_to_next_retry(output)
           wait { output.num_errors == errors + 2 }
           wait { output.retry.nil? }
@@ -1724,7 +1725,7 @@ def evacuate_dir(plugin_id)
         end
 
         assert_equal [1, 2, 2, 2], Array.new(4) { reach_retry_limit.call }
-        assert_equal 2, output.log.out.logs.count { |l| l.include?("evacuate_limit_size is reached") }
+        assert_equal 2, output.log.out.logs.count { |l| l.include?("to stay within evacuate_limit_size") }
 
         FileUtils.rm(Dir.glob(File.join(evacuate_dir(plugin_id), "*")))
         assert_equal 1, reach_retry_limit.call
diff --git a/test/plugin/test_buf_file_single.rb b/test/plugin/test_buf_file_single.rb
index ece5fddb68..91f80603b9 100644
--- a/test/plugin/test_buf_file_single.rb
+++ b/test/plugin/test_buf_file_single.rb
@@ -1240,6 +1240,7 @@ def evacuate_dir(plugin_id)
           errors = output.num_errors
           emit_events(output, tag, dummy_event_stream)
           wait { output.num_errors == errors + 1 }
+          wait { output.retry }
           proceed_to_next_retry(output)
           wait { output.num_errors == errors + 2 }
           wait { output.retry.nil? }
@@ -1247,7 +1248,7 @@ def evacuate_dir(plugin_id)
         end
 
         assert_equal [1, 2, 2, 2], Array.new(4) { reach_retry_limit.call }
-        assert_equal 2, output.log.out.logs.count { |l| l.include?("evacuate_limit_size is reached") }
+        assert_equal 2, output.log.out.logs.count { |l| l.include?("to stay within evacuate_limit_size") }
 
         FileUtils.rm(Dir.glob(File.join(evacuate_dir(plugin_id), "*")))
         assert_equal 1, reach_retry_limit.call
diff --git a/test/plugin/test_buffer.rb b/test/plugin/test_buffer.rb
index 8cb0792ad9..31f1360ed5 100644
--- a/test/plugin/test_buffer.rb
+++ b/test/plugin/test_buffer.rb
@@ -85,6 +85,11 @@ def generate_chunk(metadata)
       DummyMemoryChunk.new(metadata, compress: @compress, decompression_size_limit: @decompression_size_limit)
     end
   end
+  class DummyPersistentPlugin < DummyPlugin
+    def persistent?
+      true
+    end
+  end
 end
 
 # DummyPlugin#resume pre-seeds staged/queued chunks; tests that assert exact
@@ -96,11 +101,11 @@ def resume
 end
 
 class BufferTest < Test::Unit::TestCase
-  def create_buffer(hash)
+  def create_buffer(hash, klass: FluentPluginBufferTest::DummyPlugin)
     buffer_conf = config_element('buffer', '', hash, [])
     owner = FluentPluginBufferTest::DummyOutputPlugin.new
     owner.configure(config_element('ROOT', '', {}, [ buffer_conf ]))
-    p = FluentPluginBufferTest::DummyPlugin.new
+    p = klass.new
     p.owner = owner
     p.configure(buffer_conf)
     p
@@ -528,7 +533,7 @@ def create_chunk_es(metadata, es)
     end
 
     test '#clear_queue! skips evacuation entirely when evacuate_limit_size is 0, but still purges' do
-      p = create_buffer({'evacuate_limit_size' => 0})
+      p = create_buffer({'evacuate_limit_size' => 0}, klass: FluentPluginBufferTest::DummyPersistentPlugin)
       evacuated = []
       p.define_singleton_method(:evacuate_chunk) { |chunk| evacuated << chunk.bytesize }
       p.start
@@ -539,13 +544,14 @@ def create_chunk_es(metadata, es)
       assert_equal [], evacuated
       assert_equal [], p.queue
       assert{ qchunks.all?{ |c| c.purged } }
-      assert{ $log.out.logs.none? { |l| l.include?("evacuate_limit_size is reached") } }
+      assert{ $log.out.logs.none? { |l| l.include?("to stay within evacuate_limit_size") } }
     end
 
     test '#clear_queue! bounds the total evacuated size to evacuate_limit_size' do
-      p = create_buffer({'evacuate_limit_size' => 200})
+      p = create_buffer({'evacuate_limit_size' => 200}, klass: FluentPluginBufferTest::DummyPersistentPlugin)
       evacuated = []
       p.define_singleton_method(:evacuate_chunk) { |chunk| evacuated << chunk.bytesize }
+      p.define_singleton_method(:evacuated_bytesize) { 0 }
       p.start
       qchunks = p.queue.dup
 
@@ -560,9 +566,10 @@ def create_chunk_es(metadata, es)
     end
 
     test '#clear_queue! skips the chunks which do not fit in evacuate_limit_size' do
-      p = create_buffer({'evacuate_limit_size' => 150})
+      p = create_buffer({'evacuate_limit_size' => 150}, klass: FluentPluginBufferTest::DummyPersistentPlugin)
       evacuated = []
       p.define_singleton_method(:evacuate_chunk) { |chunk| evacuated << chunk.bytesize }
+      p.define_singleton_method(:evacuated_bytesize) { 0 }
       p.start
       qchunks = p.queue.dup
 
@@ -571,13 +578,13 @@ def create_chunk_es(metadata, es)
       assert_equal [100, 3], evacuated
       assert_equal [], p.queue
       assert{ qchunks.all?{ |c| c.purged } }
-      skipped_logs = $log.out.logs.select { |l| l.include?("evacuate_limit_size is reached") }
+      skipped_logs = $log.out.logs.select { |l| l.include?("to stay within evacuate_limit_size") }
       assert_equal 1, skipped_logs.size
       assert{ skipped_logs.first.include?("chunks=1 bytesize=100") }
     end
 
     test '#clear_queue! counts already evacuated files toward evacuate_limit_size' do
-      p = create_buffer({'evacuate_limit_size' => 200})
+      p = create_buffer({'evacuate_limit_size' => 200}, klass: FluentPluginBufferTest::DummyPersistentPlugin)
       evacuated = []
       p.define_singleton_method(:evacuate_chunk) { |chunk| evacuated << chunk.bytesize }
       p.define_singleton_method(:evacuated_bytesize) { 100 }
@@ -591,20 +598,28 @@ def create_chunk_es(metadata, es)
       assert{ qchunks.all?{ |c| c.purged } }
     end
 
-    test '#clear_queue! purges all chunks even if it fails to get the size of the evacuated files' do
-      p = create_buffer({'evacuate_limit_size' => 200})
+    test '#clear_queue! purges all chunks without evacuation if it fails to get the size of the evacuated files' do
+      p = create_buffer({'evacuate_limit_size' => 200}, klass: FluentPluginBufferTest::DummyPersistentPlugin)
       evacuated = []
       p.define_singleton_method(:evacuate_chunk) { |chunk| evacuated << chunk.bytesize }
-      p.define_singleton_method(:evacuate_dir) { raise Encoding::CompatibilityError, "incompatible character encodings" }
+      p.define_singleton_method(:evacuate_dir) { raise Errno::EACCES, "/tmp/fluent/buffer/test" }
       p.start
       qchunks = p.queue.dup
 
       p.clear_queue!
 
-      assert_equal [100, 100], evacuated
+      assert_equal [], evacuated
       assert_equal [], p.queue
       assert{ qchunks.all?{ |c| c.purged } }
       assert{ $log.out.logs.any? { |l| l.include?("failed to get the size of the evacuated chunk files") } }
+      assert{ $log.out.logs.any? { |l| l.include?("to stay within evacuate_limit_size") && l.include?("chunks=3 bytesize=203") } }
+    end
+
+    test 'ignores evacuate_limit_size if the buffer is not persistent' do
+      p = create_buffer({'evacuate_limit_size' => 200})
+
+      assert_nil p.evacuate_limit_size
+      assert{ $log.out.logs.any? { |l| l.include?("evacuate_limit_size is ignored") } }
     end
 
     test 'raises ConfigError if evacuate_limit_size is negative' do
@@ -617,15 +632,15 @@ def create_chunk_es(metadata, es)
       'without @id' => [{}, true],
       'with @id' => [{'@id' => 'test_output'}, false],
     )
-    test 'warns that evacuate_limit_size needs @id to count the files evacuated before restart' do |(owner_conf, warned)|
+    test 'warns that evacuate_limit_size should be used with @id' do |(owner_conf, warned)|
       buffer_conf = config_element('buffer', '', {'evacuate_limit_size' => '1g'})
       owner = FluentPluginBufferTest::DummyOutputPlugin.new
       owner.configure(config_element('ROOT', '', owner_conf, [buffer_conf]))
-      p = FluentPluginBufferTest::DummyPlugin.new
+      p = FluentPluginBufferTest::DummyPersistentPlugin.new
       p.owner = owner
       p.configure(buffer_conf)
 
-      assert_equal warned, $log.out.logs.any? { |l| l.include?("unless @id is configured") }
+      assert_equal warned, $log.out.logs.any? { |l| l.include?("should be used with @id") }
     end
 
     data(
@@ -633,7 +648,7 @@ def create_chunk_es(metadata, es)
       'larger than chunk_limit_size' => ['1g', false],
     )
     test 'warns if evacuate_limit_size is smaller than chunk_limit_size' do |(limit, warned)|
-      create_buffer({'evacuate_limit_size' => limit, 'chunk_limit_size' => '8m'})
+      create_buffer({'evacuate_limit_size' => limit, 'chunk_limit_size' => '8m'}, klass: FluentPluginBufferTest::DummyPersistentPlugin)
 
       assert_equal warned, $log.out.logs.any? { |l| l.include?("smaller than chunk_limit_size") }
     end

From bf4e37f45099c2d36ffc81390cfe3f11cf584d23 Mon Sep 17 00:00:00 2001
From: Shizuo Fujita <fujita@clear-code.com>
Date: Tue, 20 Oct 2026 11:13:55 +0900
Subject: [PATCH 5/6] buffer: describe evacuate_limit_size as an approximate
 limit

The total size may exceed the limit due to chunk metadata or concurrent
evacuation by multiple workers.

Signed-off-by: Shizuo Fujita <fujita@clear-code.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 lib/fluent/plugin/buffer.rb | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/lib/fluent/plugin/buffer.rb b/lib/fluent/plugin/buffer.rb
index 9bd980dff4..cc96560027 100644
--- a/lib/fluent/plugin/buffer.rb
+++ b/lib/fluent/plugin/buffer.rb
@@ -72,7 +72,7 @@ class BufferChunkOverflowError < BufferError; end # A record size is larger than
       desc 'If true, chunks are thrown away when unrecoverable error happens'
       config_param :disable_chunk_backup, :bool, default: false
 
-      desc 'The size limit of the evacuated chunk files in the evacuation directory. When the retry limit is reached, queued chunks which fit in the remaining size are evacuated, and the others are purged without being evacuated. Set 0 to disable evacuation. nil (default) means no limit.'
+      desc 'The approximate size limit of the evacuated chunk files in the evacuation directory. When the retry limit is reached, queued chunks that fit within the remaining capacity are evacuated, and the others are purged without being evacuated. The total size may exceed this limit due to chunk metadata or concurrent evacuation by multiple workers. Set to 0 to disable evacuation. nil (the default) means no limit.'
       config_param :evacuate_limit_size, :size, default: nil
 
       Metadata = Struct.new(:timekey, :tag, :variables, :seq) do

From 48a1886ce376d8846e3b3042fd23f30127dacde1 Mon Sep 17 00:00:00 2001
From: Shizuo Fujita <fujita@clear-code.com>
Date: Tue, 20 Oct 2026 11:58:24 +0900
Subject: [PATCH 6/6] buffer: mention reload in the warning about @id for
 evacuate_limit_size

Without @id, a graceful reload also changes the plugin_id, and then the
files evacuated before are not counted.

Signed-off-by: Shizuo Fujita <fujita@clear-code.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 lib/fluent/plugin/buffer.rb | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/lib/fluent/plugin/buffer.rb b/lib/fluent/plugin/buffer.rb
index cc96560027..6624c9c2e4 100644
--- a/lib/fluent/plugin/buffer.rb
+++ b/lib/fluent/plugin/buffer.rb
@@ -237,7 +237,7 @@ def configure(conf)
             @evacuate_limit_size = nil
           elsif @evacuate_limit_size > 0
             unless owner.plugin_id_configured?
-              log.warn "evacuate_limit_size should be used with @id, because the evacuation directory without @id can change when the configuration is changed"
+              log.warn "evacuate_limit_size should be used with @id, because the evacuation directory without @id can change when the configuration is changed or reloaded"
             end
             if @evacuate_limit_size < @chunk_limit_size
               log.warn "evacuate_limit_size is smaller than chunk_limit_size, so full chunks may not be evacuated", evacuate_limit_size: @evacuate_limit_size, chunk_limit_size: @chunk_limit_size