buffer: fix an enqueued unstaged file chunk being re-staged and flushed twice

FileChunk#enqueued! and FileSingleChunk#enqueued! did nothing for an
unstaged chunk, so a chunk pushed by Buffer#enqueue_unstaged_chunk stayed
:unstaged with its staged file name while it was in the queue. When
write_step_by_step unstages chunks on a single-record overflow,
Buffer#write re-staged that queued chunk because the `u.unstaged?` guard
could not tell it was already enqueued, and the same chunk object lived
in both @Queue and @stage. The second flush of that object raised
"closed stream" at FileChunk#open and ENOENT at FileChunk#purge, and the
buffer size gauges leaked.

Treat an unstaged chunk like a staged one in enqueued!: mark it as
:queued, write its metadata and rename its files to the queued path, so
resume also restores it as a queued chunk. The state changes before the
rename and the gauges are updated before enqueued!, so a chunk whose
rename fails stays a consistent queued chunk with its staged file name.
Buffer#write only warns about such a failure and keeps enqueueing the
other chunks: the records are already queued, and a chunk left unstaged
there would be purged with its committed records.

FileSingleBuffer#resume now enqueues a staged chunk file whose metadata
is already staged, as FileBuffer#resume does, so that two chunk files
with the staged name are both restored. file_rename reopens the old file
when the rename fails on Windows, so that the chunk stays readable.

Fixes #4662

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Signed-off-by: Shizuo Fujita <fujita@clear-code.com>

Author: Watson1978

lib/fluent/plugin/buf_file.rb

@@ -174,7 +174,7 @@ def resume
           when :staged
             # unstaged chunk created at Buffer#write_step_by_step is identified as the staged chunk here because FileChunk#assume_chunk_state checks only the file name.
             # https://github.com/fluent/fluentd/blob/9d113029d4550ce576d8825bfa9612aa3e55bff0/lib/fluent/plugin/buffer.rb#L663
-            # This case can happen when fluentd process is killed by signal or other reasons between creating unstaged chunks and changing them to staged mode in Buffer#write
+            # This case can happen when fluentd process is killed by signal or other reasons between creating unstaged chunks and enqueueing them (which renames them) in Buffer#write
             # these chunks(unstaged chunks) has shared the same metadata
             # So perform enqueue step again https://github.com/fluent/fluentd/blob/9d113029d4550ce576d8825bfa9612aa3e55bff0/lib/fluent/plugin/buffer.rb#L364
             if chunk_size_full?(chunk) || stage.key?(chunk.metadata)

lib/fluent/plugin/buf_file_single.rb

@@ -193,7 +193,11 @@ def resume
 
           case chunk.state
           when :staged
-            stage[chunk.metadata] = chunk
+            if chunk_size_full?(chunk) || stage.key?(chunk.metadata)
+              queue << chunk.enqueued!
+            else
+              stage[chunk.metadata] = chunk
+            end
           when :queued
             queue << chunk
           end

lib/fluent/plugin/buffer.rb

@@ -418,31 +418,41 @@ def write(metadata_and_data, format: nil, size: nil, enqueue: false)
             end
           end
 
+          enqueue_errors = []
           chunks_to_enqueue.each do |c|
-            if c.staged? && (enqueue || chunk_size_full?(c))
-              m = c.metadata
-              enqueue_chunk(m)
-              if unstaged_chunks[m] && !unstaged_chunks[m].empty?
-                u = unstaged_chunks[m].pop
-                u.synchronize do
-                  if u.unstaged? && !chunk_size_full?(u)
-                    # `u.metadata.seq` and `m.seq` can be different but Buffer#enqueue_chunk expect them to be the same value
-                    u.metadata.seq = 0
-                    synchronize {
-                      @stage[m] = u.staged!
-                      @stage_size_metrics.add(u.bytesize)
-                    }
+            begin
+              if c.staged? && (enqueue || chunk_size_full?(c))
+                m = c.metadata
+                enqueue_chunk(m)
+                if unstaged_chunks[m] && !unstaged_chunks[m].empty?
+                  u = unstaged_chunks[m].pop
+                  u.synchronize do
+                    if u.unstaged? && !chunk_size_full?(u)
+                      # `u.metadata.seq` and `m.seq` can be different but Buffer#enqueue_chunk expect them to be the same value
+                      u.metadata.seq = 0
+                      synchronize {
+                        @stage[m] = u.staged!
+                        @stage_size_metrics.add(u.bytesize)
+                      }
+                    end
                   end
                 end
+              elsif c.unstaged?
+                enqueue_unstaged_chunk(c)
+              else
+                # previously staged chunk is already enqueued, closed or purged.
+                # no problem.
               end
-            elsif c.unstaged?
-              enqueue_unstaged_chunk(c)
-            else
-              # previously staged chunk is already enqueued, closed or purged.
-              # no problem.
+            rescue => e
+              enqueue_errors << e
             end
           end
 
+          enqueue_errors.each do |e|
+            log.warn "error occurs in enqueueing a chunk", error: e
+            log.warn_backtrace e.backtrace
+          end
+
           operated_chunks.clear if errors.empty?
 
           if errors.size > 0
@@ -492,6 +502,9 @@ def enqueue_chunk(metadata)
 
         chunk.synchronize do
           synchronize do
+            bytesize = chunk.bytesize
+            @stage_size_metrics.sub(bytesize)
+            @queue_size_metrics.add(bytesize)
             if chunk.empty?
               chunk.close
             else
@@ -500,9 +513,6 @@ def enqueue_chunk(metadata)
               @queued_num[metadata] = @queued_num.fetch(metadata, 0) + 1
               chunk.enqueued!
             end
-            bytesize = chunk.bytesize
-            @stage_size_metrics.sub(bytesize)
-            @queue_size_metrics.add(bytesize)
           end
         end
         nil
@@ -517,9 +527,9 @@ def enqueue_unstaged_chunk(chunk)
             metadata.seq = 0 # metadata.seq should be 0 for counting @queued_num
             @queue << chunk
             @queued_num[metadata] = @queued_num.fetch(metadata, 0) + 1
+            @queue_size_metrics.add(chunk.bytesize)
             chunk.enqueued!
           end
-          @queue_size_metrics.add(chunk.bytesize)
         end
       end
 

lib/fluent/plugin/buffer/file_chunk.rb

@@ -100,8 +100,14 @@ def empty?
         end
 
         def enqueued!
-          return unless self.staged?
+          return self unless self.writable?
 
+          super
+          rename_to_queued_path
+          self
+        end
+
+        def rename_to_queued_path
           new_chunk_path = self.class.generate_queued_chunk_path(@path, @unique_id)
           new_meta_path = new_chunk_path + '.meta'
 
@@ -139,9 +145,8 @@ def enqueued!
 
           @path = new_chunk_path
           @meta_path = new_meta_path
-
-          super
         end
+        private :rename_to_queued_path
 
         def close
           super
@@ -266,19 +271,27 @@ def write_metadata(update: true)
 
         def file_rename(file, old_path, new_path, callback=nil)
           pos = file.pos
+          setup = ->(f) {
+            f.set_encoding(Encoding::ASCII_8BIT)
+            f.sync = true
+            f.binmode
+            f.pos = pos
+            callback.call(f) if callback
+          }
           if Fluent.windows?
             file.close
-            File.rename(old_path, new_path)
-            file = File.open(new_path, 'rb', @permission)
+            renamed = false
+            begin
+              File.rename(old_path, new_path)
+              renamed = true
+            ensure
+              setup.call(File.open(renamed ? new_path : old_path, 'rb', @permission))
+            end
           else
             File.rename(old_path, new_path)
             file.reopen(new_path, 'rb')
+            setup.call(file)
           end
-          file.set_encoding(Encoding::ASCII_8BIT)
-          file.sync = true
-          file.binmode
-          file.pos = pos
-          callback.call(file) if callback
         end
 
         def create_new_chunk(path, perm)

lib/fluent/plugin/buffer/file_single_chunk.rb

@@ -93,8 +93,14 @@ def empty?
         end
 
         def enqueued!
-          return unless self.staged?
+          return self unless self.writable?
 
+          super
+          rename_to_queued_path
+          self
+        end
+
+        def rename_to_queued_path
           new_chunk_path = self.class.generate_queued_chunk_path(@path, @unique_id)
 
           begin
@@ -114,9 +120,8 @@ def enqueued!
           end
 
           @path = new_chunk_path
-
-          super
         end
+        private :rename_to_queued_path
 
         def close
           super
@@ -225,19 +230,27 @@ def restore_size(chunk_format)
 
         def file_rename(file, old_path, new_path, callback = nil)
           pos = file.pos
+          setup = ->(f) {
+            f.set_encoding(Encoding::ASCII_8BIT)
+            f.sync = true
+            f.binmode
+            f.pos = pos
+            callback.call(f) if callback
+          }
           if Fluent.windows?
             file.close
-            File.rename(old_path, new_path)
-            file = File.open(new_path, 'rb', @permission)
+            renamed = false
+            begin
+              File.rename(old_path, new_path)
+              renamed = true
+            ensure
+              setup.call(File.open(renamed ? new_path : old_path, 'rb', @permission))
+            end
           else
             File.rename(old_path, new_path)
             file.reopen(new_path, 'rb')
+            setup.call(file)
           end
-          file.set_encoding(Encoding::ASCII_8BIT)
-          file.sync = true
-          file.binmode
-          file.pos = pos
-          callback.call(file) if callback
         end
 
         ESCAPE_REGEXP = /[^-_.a-zA-Z0-9]/n

test/plugin/test_buf_file.rb

@@ -444,6 +444,145 @@ def write_metadata(path, chunk_id, metadata, size, ctime, mtime)
     end
   end
 
+  sub_test_case 'unstaged chunks created by write_step_by_step' do
+    setup do
+      @bufdir = File.expand_path('../../tmp/buffer_file_unstaged', __FILE__)
+      @bufpath = File.join(@bufdir, 'testbuf.*.log')
+      FileUtils.rm_r @bufdir if File.exist?(@bufdir)
+
+      Fluent::Test.setup
+      @d = FluentPluginFileBufferTest::DummyOutputPlugin.new
+      @p = Fluent::Plugin::FileBuffer.new
+      @p.owner = @d
+      @p.configure(config_element('buffer', '', {'path' => @bufpath, 'chunk_limit_size' => 1024}))
+      @p.start
+    end
+
+    teardown do
+      if @p
+        @p.stop unless @p.stopped?
+        @p.before_shutdown unless @p.before_shutdown?
+        @p.shutdown unless @p.shutdown?
+        @p.after_shutdown unless @p.after_shutdown?
+        @p.close unless @p.closed?
+        @p.terminate unless @p.terminated?
+      end
+      FileUtils.rm_r @bufdir if File.exist?(@bufdir)
+    end
+
+    def write_with_single_record_overflow(m)
+      @p.write({m => ["a" * 600]})
+      @p.write({m => ["b" * 600, "c" * 600, "d" * 380]})
+    end
+
+    test 'enqueued unstaged chunks are queued and are not re-staged' do
+      m = @p.metadata
+      write_with_single_record_overflow(m)
+
+      assert_equal [:queued, :queued, :queued], @p.queue.map(&:state)
+      assert_equal [600, 600, 980], @p.queue.map(&:bytesize)
+      assert_true @p.queue.all? { |c| File.basename(c.path).start_with?('testbuf.q') }
+      assert_true @p.stage.empty?
+      assert_equal 0, @p.stage_size
+      assert_equal 2180, @p.queue_size
+    end
+
+    test 'enqueued unstaged chunks are resumed as queued chunks' do
+      m = @p.metadata
+      write_with_single_record_overflow(m)
+      @p.stop; @p.before_shutdown; @p.shutdown; @p.after_shutdown; @p.close; @p.terminate
+
+      @p = Fluent::Plugin::FileBuffer.new
+      @p.owner = @d
+      @p.configure(config_element('buffer', '', {'path' => @bufpath, 'chunk_limit_size' => 1024}))
+      @p.start
+
+      assert_true @p.stage.empty?
+      assert_equal [600, 600, 980], @p.queue.map(&:bytesize).sort
+      assert_equal 0, @p.stage_size
+      assert_equal 2180, @p.queue_size
+    end
+
+    test 'committed data is kept when renaming an enqueued unstaged chunk fails' do
+      m = @p.metadata
+      @p.write({m => ["a" * 600]})
+      first = @p.stage[m]
+      first.define_singleton_method(:file_rename) { |*| raise Errno::EACCES, "injected" }
+
+      assert_nothing_raised { @p.write({m => ["b" * 600, "c" * 600, "d" * 380]}) }
+      assert_equal 1, $log.out.logs.count { |l| l.include?("error occurs in enqueueing a chunk") }
+
+      assert_true @p.queue.any? { |c| c.equal?(first) }
+      assert_equal :queued, first.state
+      assert_equal 600, first.bytesize
+      assert_true File.exist?(first.path)
+      assert_equal @p.queue.sum(&:bytesize), @p.queue_size
+
+      read = +""
+      while (chunk = @p.dequeue_chunk)
+        assert_nothing_raised { read << chunk.read }
+        @p.purge_chunk(chunk.unique_id)
+      end
+      assert_true read.include?("a" * 600)
+      assert_true read.include?("b" * 600)
+      assert_equal 0, @p.queue_size
+    end
+
+    test 'all data is resumed after a rename failure and a restart' do
+      m = @p.metadata
+      @p.write({m => ["a" * 600]})
+      first = @p.stage[m]
+      first.define_singleton_method(:file_rename) { |*| raise Errno::EACCES, "injected" }
+      @p.write({m => ["b" * 600, "c" * 600, "d" * 380]})
+      @p.write({m => ["e" * 100]})
+      @p.stop; @p.before_shutdown; @p.shutdown; @p.after_shutdown; @p.close; @p.terminate
+
+      @p = Fluent::Plugin::FileBuffer.new
+      @p.owner = @d
+      @p.configure(config_element('buffer', '', {'path' => @bufpath, 'chunk_limit_size' => 1024}))
+      @p.start
+
+      chunks = @p.stage.values + @p.queue
+      assert_equal 4, chunks.size
+      assert_equal 600 + 1580 + 100, chunks.sum(&:bytesize)
+      assert_equal chunks.sum(&:bytesize), @p.stage_size + @p.queue_size
+    end
+
+    test 'a staged chunk stays queued with consistent gauges when its rename fails' do
+      m = @p.metadata
+      @p.write({m => ["a" * 600]})
+      chunk = @p.stage[m]
+      chunk.define_singleton_method(:file_rename) { |*| raise Errno::EACCES, "injected" }
+
+      assert_raise(RuntimeError) { @p.enqueue_chunk(m) }
+
+      assert_true @p.stage.empty?
+      assert_equal [chunk], @p.queue
+      assert_equal :queued, chunk.state
+      assert_equal 0, @p.stage_size
+      assert_equal 600, @p.queue_size
+
+      assert_nothing_raised { @p.dequeue_chunk.read }
+      @p.purge_chunk(chunk.unique_id)
+      assert_equal 0, @p.queue_size
+    end
+
+    test 'each chunk is flushed once and the gauges return to zero' do
+      m = @p.metadata
+      write_with_single_record_overflow(m)
+      @p.write({m => ["e" * 600, "f" * 300]})
+      assert_equal @p.queue.size, @p.queue.map(&:object_id).uniq.size
+
+      while (chunk = @p.dequeue_chunk)
+        assert_nothing_raised { chunk.read }
+        @p.purge_chunk(chunk.unique_id)
+      end
+
+      assert_equal 0, @p.queue_size
+      assert_equal @p.stage.values.sum(&:bytesize), @p.stage_size
+    end
+  end
+
   sub_test_case 'configured with system root directory and plugin @id' do
     setup do
       @root_dir = File.expand_path('../../tmp/buffer_file_root', __FILE__)