fix(out_file): create symlink for every latest metadata sharing a timekey

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

lib/fluent/plugin/out_file.rb

@@ -18,6 +18,7 @@
 require 'zlib'
 require 'time'
 require 'pathname'
+require 'thread'
 
 require 'fluent/plugin/output'
 require 'fluent/config/error'
@@ -75,9 +76,30 @@ module SymlinkBufferMixin
       def metadata(timekey: nil, tag: nil, variables: nil)
         metadata = super
 
-        @latest_metadata ||= new_metadata(timekey: 0)
-        if metadata.timekey && (metadata.timekey >= @latest_metadata.timekey)
-          @latest_metadata = metadata
+        @symlink_mutex ||= Mutex.new
+        @symlink_mutex.synchronize do
+          @latest_metadata ||= new_metadata(timekey: 0)
+          if metadata.timekey && (metadata.timekey >= @latest_metadata.timekey)
+            if metadata.timekey == @latest_metadata.timekey
+              # Same timekey, different variables (e.g. bulk input with several
+              # chunk-key values): every one of these is "latest" for its own
+              # chunk and must get a symlink (fluentd#5099).
+              #
+              # NOTE: Metadata#hash only hashes timekey (by design, for perf), so
+              # keying a Hash on Metadata objects makes every distinct metadata
+              # sharing a timekey collide into one bucket -> O(n^2). Use a
+              # compound key [timekey, tag, variables] that uniquely identifies
+              # the metadata's chunk-key identity instead.
+              key = [metadata.timekey, metadata.tag, metadata.variables]
+              (@latest_metadatas ||= {})[key] = metadata
+              @latest_metadata = metadata
+            else
+              # Timekey advanced: drop the previous window's tracking entirely
+              # so memory does not grow without bound within the new timekey.
+              @latest_metadata = metadata
+              @latest_metadatas = { [metadata.timekey, metadata.tag, metadata.variables] => metadata }
+            end
+          end
         end
 
         metadata
@@ -93,11 +115,24 @@ def symlink_path=(path)
 
       def generate_chunk(metadata)
         chunk = super
-        # "symlink" feature is to link from symlink_path to the latest file chunk. Records with latest
-        # timekey will be appended into that file chunk. On the other side, resumed file chunks might NOT
-        # have timekey, especially in the cases that resumed file chunks are generated by Fluentd v0.12.
-        # These chunks will be enqueued immediately, and will be flushed soon.
-        if chunk.metadata == @latest_metadata
+        # Snapshot the tracked set under the same lock used by #metadata so we
+        # never read a half-updated Hash (lost-update race flagged in review).
+        latest_set = nil
+        @symlink_mutex ||= Mutex.new
+        @symlink_mutex.synchronize do
+          latest_set = @latest_metadatas || { [@latest_metadata.timekey, @latest_metadata.tag, @latest_metadata.variables] => @latest_metadata }
+        end
+        # "symlink" feature links symlink_path to the latest file chunk. Records
+        # with the latest timekey are appended into that chunk. Resumed file
+        # chunks (e.g. from Fluentd v0.12) may have no timekey and are flushed
+        # immediately, so they are intentionally excluded here.
+        #
+        # The former `chunk.metadata == @latest_metadata` fallback was dead code:
+        # whenever @latest_metadatas is non-nil, @latest_metadata is always one
+        # of its values (assigned together above), and when it is nil the
+        # fallback builds the same single-entry set inline.
+        key = [chunk.metadata.timekey, chunk.metadata.tag, chunk.metadata.variables]
+        if latest_set.key?(key)
           sym_path = @_output_plugin_for_symlink.extract_placeholders(@_symlink_path, chunk)
           FileUtils.mkdir_p(File.dirname(sym_path), mode: @_output_plugin_for_symlink.dir_perm)
           if @_output_plugin_for_symlink.symlink_path_use_relative

test/plugin/test_buffer.rb

@@ -435,6 +435,32 @@ def create_chunk_es(metadata, es)
       assert_equal({}, @p.dequeued)
     end
 
+    test '#purge_chunk releases queue_size even when chunk.purge raises (#5468)' do
+      # Simulate a physical purge failure (unlink/close on the buffer path)
+      # so chunk.purge raises after the chunk is dequeued.
+      failing_purge = Module.new do
+        def purge
+          raise IOError, 'simulated purge failure (unlink EIO)'
+        end
+      end
+      @p.buffer_class::Chunk.include(failing_purge)
+
+      m1 = @p.dequeue_chunk
+      assert_equal [@dm0, @dm1, @dm1], @p.queue.map(&:metadata)
+      assert_equal({m1.unique_id => m1}, @p.dequeued)
+
+      queued_before = @p.queue_size
+      assert queued_before > 0
+
+      # purge_chunk swallows the error but must still decrement queue_size
+      @p.purge_chunk(m1.unique_id)
+
+      assert m1.purged
+      # queue_size must return to its pre-dequeue value, not leak upward
+      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)

test/plugin/test_out_file.rb

@@ -779,6 +779,38 @@ def parse_system(text)
       end
     end
 
+    test 'symlink with placeholders and variable chunk keys creates a symlink per metadata (bulk input)' do
+      omit "Windows doesn't support symlink" if Fluent.windows?
+      conf = %[
+        path #{TMP_DIR}/${tag}/out_file_${myid}_%Y%m%d.log
+        symlink_path #{SYMLINK_PATH}/foo/${tag}_${myid}.log
+        <buffer tag,time,myid>
+        </buffer>
+      ]
+
+      d = create_driver(conf)
+      begin
+        d.run(default_tag: 'tag') do
+          # Bulk input: two events with different `myid` values but the SAME
+          # timekey arrive in one stream. Each metadata must get its own
+          # staged chunk and its own symlink (fluentd#5099).
+          es = Fluent::MultiEventStream.new
+          t = event_time("2011-01-02 13:14:15 UTC")
+          es.add(t, {"a" => 1, "myid" => "1"})
+          es.add(t, {"a" => 2, "myid" => "2"})
+          d.feed(es)
+        end
+
+        assert File.symlink?("#{SYMLINK_PATH}/foo/tag_1.log"),
+               "missing symlink for myid=1"
+        assert File.symlink?("#{SYMLINK_PATH}/foo/tag_2.log"),
+               "missing symlink for myid=2"
+      ensure
+        FileUtils.rm_f("#{SYMLINK_PATH}/foo/tag_1.log")
+        FileUtils.rm_f("#{SYMLINK_PATH}/foo/tag_2.log")
+      end
+    end
+
     test 'relative symlink' do
       omit "Windows doesn't support symlinks" if Fluent.windows?