file_chunk: add stricter checks for broken meta files (#4998)

**Which issue(s) this PR fixes**: 
* Related to #3970

**What this PR does / why we need it**: 
This PR improves meta file corruption checking.

The meta file contains at least the following field values.


https://github.com/fluent/fluentd/blob/fa2eb58922e1c36f83bf1d5243b325a860f72864/lib/fluent/plugin/buffer/file_chunk.rb#L249-L254

This PR reinforces #1874.


Without this changes, it might causes following error when launch
fluentd every time with broken meta file:

```
2025-06-06 12:11:26 +0900 [error]: unexpected error while checking flushed chunks. ignored. error_class=NoMethodError error="undefined method '<' for nil"
  2025-06-06 12:11:26 +0900 [error]: /Users/watson/src/fluentd/lib/fluent/plugin/output.rb:1479:in 'block in Fluent::Plugin::Output#enqueue_thread_run'
  2025-06-06 12:11:26 +0900 [error]: /Users/watson/src/fluentd/lib/fluent/plugin/buffer.rb:548:in 'block in Fluent::Plugin::Buffer#enqueue_all'
  2025-06-06 12:11:26 +0900 [error]: /Users/watson/src/fluentd/lib/fluent/plugin/buffer.rb:542:in 'Array#each'
  2025-06-06 12:11:26 +0900 [error]: /Users/watson/src/fluentd/lib/fluent/plugin/buffer.rb:542:in 'Fluent::Plugin::Buffer#enqueue_all'
  2025-06-06 12:11:26 +0900 [error]: /Users/watson/src/fluentd/lib/fluent/plugin/output.rb:1479:in 'Fluent::Plugin::Output#enqueue_thread_run'
  2025-06-06 12:11:26 +0900 [error]: /Users/watson/src/fluentd/lib/fluent/plugin_helper/thread.rb:78:in 'block in Fluent::PluginHelper::Thread#thread_create'
```

If the timekey value is corrupted, the above error occurs.
Since there is no appropriate way to check timekey directly, check `id`,
`c`, and `m` fields instead. This is because when timekey is broken,
other fields may also be broken.

It might be possible that the `@size` is 0.
`@unique_id`, `@created_at`, and `@modified_at` are set when FileChunk
is initialized, so they definitely have some values.
I think these fields should be written in meta file.

So, this PR adds the `id`, `c`, and `m` fields check.


Previously, it operates using default value if metadata was broken. 
However, it can miss the corruption and result in unexpected errors.
So, this PR enhances the detection of broken metadata files instead of
using defalut value.

This change has backward compatible with v0.14 behavior.


**Docs Changes**:
Not necessarily required.

**Release Note**: 
buf_file: reinforce buffer file corruption check

---------

Signed-off-by: Shizuo Fujita <fujita@clear-code.com>

Author: Watson1978

lib/fluent/plugin/buffer/file_chunk.rb

@@ -219,13 +219,17 @@ def restore_metadata(bindata)
             # old type of restore
             data = Fluent::MessagePackFactory.msgpack_unpacker(symbolize_keys: true).feed(bindata).read rescue {}
           end
+          raise FileChunkError, "invalid meta data" if data.nil? || !data.is_a?(Hash)
+          raise FileChunkError, "invalid unique_id" unless data[:id]
+          raise FileChunkError, "invalid created_at" unless data[:c].to_i > 0
+          raise FileChunkError, "invalid modified_at" unless data[:m].to_i > 0
 
           now = Fluent::Clock.real_now
 
-          @unique_id = data[:id] || self.class.unique_id_from_path(@path) || @unique_id
+          @unique_id = data[:id]
           @size = data[:s] || 0
-          @created_at = data.fetch(:c, now.to_i)
-          @modified_at = data.fetch(:m, now.to_i)
+          @created_at = data[:c]
+          @modified_at = data[:m]
 
           @metadata.timekey = data[:timekey]
           @metadata.tag = data[:tag]

test/plugin/test_buf_file.rb

@@ -1322,6 +1322,36 @@ def compare_log(plugin, msg)
       assert { File.exist?("#{@bufdir}/backup/worker0/#{@id_output}/#{@d.dump_unique_id_hex(c2id)}.log") }
     end
 
+    test '#resume backups enqueued broken metadata which has broken id, c, m fields' do
+      setup_plugins({'path' => @bufpath})
+      cid, path = create_first_chunk('q')
+      metadata = File.read(path + '.meta')
+      File.open(path + '.meta', 'wb') { |f| f.write(metadata[0..6] + "\0" * (metadata.size - 6)) } # create enqueued broken meta file
+
+      Fluent::SystemConfig.overwrite_system_config('root_dir' => @bufdir) do
+        @p.start
+      end
+
+      compare_log(@p, 'enqueued meta file is broken')
+      assert { not File.exist?(path) }
+      assert { File.exist?("#{@bufdir}/backup/worker0/#{@id_output}/#{@d.dump_unique_id_hex(cid)}.log") }
+    end
+
+    test '#resume backups enqueued broken metadata by truncated' do
+      setup_plugins({'path' => @bufpath})
+      cid, path = create_first_chunk('q')
+      metadata = File.read(path + '.meta')
+      File.open(path + '.meta', 'wb') { |f| f.write(metadata[0..-2]) } # create enqueued broken meta file with last byte truncated
+
+      Fluent::SystemConfig.overwrite_system_config('root_dir' => @bufdir) do
+        @p.start
+      end
+
+      compare_log(@p, 'enqueued meta file is broken')
+      assert { not File.exist?(path) }
+      assert { File.exist?("#{@bufdir}/backup/worker0/#{@id_output}/#{@d.dump_unique_id_hex(cid)}.log") }
+    end
+
     test '#resume throws away broken chunk with disable_chunk_backup' do
       setup_plugins({'path' => @bufpath, 'disable_chunk_backup' => true})
       c1id, _ = create_first_chunk('b')