From 7e7340671491c9d42d4cf81ccdccef76975e2888 Mon Sep 17 00:00:00 2001
From: Shizuo Fujita <fujita@clear-code.com>
Date: Wed, 16 Jul 2025 19:00:29 +0900
Subject: [PATCH] server plugin helper: ensure to close all connections at
 shutdown (#5026)

**Which issue(s) this PR fixes**:
Fixes #

**What this PR does / why we need it**:
TCP server with `server` helper does not close all connections at
shutdown process.
When receiving data from multiple clients, the server receive the data
continuously because the connection is not closed.

The server will shut down properly by this PR

### Reproduce
1. Launch Fluentd with following config file.
2. Send syslog data from two or more clients using following client
script
3. Terminate Fluentd
4. Relaunch Fluentd, then it shows the `2025-07-16 14:12:26 +0900
[warn]: #0 restoring buffer file: path = xxxxxxxxx` in logs.

* config

```
<source>
  @type syslog
  tag system

  <transport tcp>
  </transport>

  bind 0.0.0.0
  port 5140
</source>

<match **>
  @type file
  path "#{File.expand_path('~/tmp/fluentd/maillog')}"
  <buffer>
    @type file
    path "#{File.expand_path('~/tmp/fluentd/buffer/buffer_syslog_maillog')}"
    flush_at_shutdown true
  </buffer>
</match>
```

* client script

```ruby
require 'bundler/inline'
gemfile do
  source 'https://rubygems.org'
  gem 'remote_syslog_sender'
end

def create_client
  Thread.new do
    sender = RemoteSyslogSender.new('127.0.0.1', 5140, protocol: :tcp)
    loop do
      sender.transmit("message body")
      sleep 0.5
    end
  end
end

clients = []
3.times do
  clients << create_client
end

clients.each(&:join)
```

Example code.
```
@ary = [1,2,3]

# It would like 1, 2, and 3 to be processed.
# However, following code handles 1 and 3.
@ary.each do |i|
  puts i
  @ary.delete(i)
end
```

**Docs Changes**:
Not needed.

**Release Note**:
Same as the title.

Signed-off-by: Shizuo Fujita <fujita@clear-code.com>
Signed-off-by: Daijiro Fukuda <fukuda@clear-code.com>
---
 lib/fluent/plugin_helper/server.rb |  5 ++++-
 test/plugin_helper/test_server.rb  | 19 +++++++++++++++++++
 2 files changed, 23 insertions(+), 1 deletion(-)

diff --git a/lib/fluent/plugin_helper/server.rb b/lib/fluent/plugin_helper/server.rb
index eecee5ae7e..0f185c2512 100644
--- a/lib/fluent/plugin_helper/server.rb
+++ b/lib/fluent/plugin_helper/server.rb
@@ -347,7 +347,10 @@ def stop
       end
 
       def shutdown
-        @_server_connections.each do |conn|
+        # When it invokes conn.cose, it reduces elements in @_server_connections by close_callback,
+        # and it reduces the number of loops. This prevents the connection closing.
+        # So, it requires invoking #dup to avoid the problem.
+        @_server_connections.dup.each do |conn|
           conn.close rescue nil
         end
 
diff --git a/test/plugin_helper/test_server.rb b/test/plugin_helper/test_server.rb
index d1f3fa3319..317f03b021 100644
--- a/test/plugin_helper/test_server.rb
+++ b/test/plugin_helper/test_server.rb
@@ -364,6 +364,25 @@ class Dummy < Fluent::Plugin::TestBase
         d2.stop; d2.before_shutdown; d2.shutdown; d2.after_shutdown; d2.close; d2.terminate
       end
     end
+
+    test 'close all connections by shutdown' do
+      @d.server_create_tcp(:s, @port) do |data, conn|
+      end
+
+      client_sockets = []
+      5.times do
+        client_sockets << TCPSocket.open("127.0.0.1", @port)
+      end
+      waiting(4){ sleep 0.1 until @d.instance_variable_get(:@_server_connections).size == 5 }
+
+      @d.stop
+      @d.before_shutdown
+      @d.shutdown
+
+      assert_true @d.instance_variable_get(:@_server_connections).empty?
+    ensure
+      client_sockets.each(&:close)
+    end
   end
 
   sub_test_case '#server_create' do