CI: Fails "raises no error when another thread closes a socket"

[Watson1978]

Describe the bug

Sometimes, "raises no error when another thread closes a socket" in test_ack_handler.rb will be failed

To Reproduce

Run tests repeatedly using rake tasks like the following.
It may be easier to reproduce with the older Ruby.

task :default do
  100.times do |i|
    puts "******* test #{i} *******"
    sh 'bundle exec ruby -w -I"lib:test" -Eascii-8bit:ascii-8bit test/plugin/out_forward/test_ack_handler.rb -v --name "test: raises no error when another thread closes a socket"'
    if $?.exitstatus != 0
      exit
    end
    puts "\n\n\n\n\n"
  end
end

$ rake -f runner.rake
--- (snip) ---

******* test 20 *******
bundle exec ruby -w -I"lib:test" -Eascii-8bit:ascii-8bit test/plugin/out_forward/test_ack_handler.rb -v --name "test: raises no error when another thread closes a socket"
/Users/watson/.rbenv/versions/3.1.6/lib/ruby/gems/3.1.0/gems/async-2.18.0/lib/async/task.rb:290: warning: assigned but unused variable - error
/Users/watson/.rbenv/versions/3.1.6/lib/ruby/gems/3.1.0/gems/async-pool-0.10.1/lib/async/pool/controller.rb:336: warning: assigned but unused variable - error
/Users/watson/.rbenv/versions/3.1.6/lib/ruby/gems/3.1.0/gems/async-http-0.82.2/lib/async/http/protocol/http1/server.rb:55: warning: assigned but unused variable - error
/Users/watson/.rbenv/versions/3.1.6/lib/ruby/gems/3.1.0/gems/async-http-0.82.2/lib/async/http/protocol/http2/connection.rb:98: warning: assigned but unused variable - ignored_error
/Users/watson/.rbenv/versions/3.1.6/lib/ruby/gems/3.1.0/gems/protocol-http-0.42.0/lib/protocol/http/body/stream.rb:233: warning: assigned but unused variable - buffer
/Users/watson/src/fluentd/lib/fluent/plugin_helper.rb:46: warning: method redefined; discarding old inherited
/Users/watson/src/fluentd/lib/fluent/plugin_helper.rb:46: warning: previous definition of inherited was here
Loaded suite test/plugin/out_forward/test_ack_handler
Started
AckHandlerTest:
  test: raises no error when another thread closes a socket:																		F
=============================================================================================================================================================================================================================
Failure: test: raises no error when another thread closes a socket(AckHandlerTest):
  2024-11-03 13:48:36 +0900 [error]: unexpected error while receiving ack message error_class=NameError error="method `recv' not defined in #<Class:#<IO:0x00000001052069e8>>"
  \n  2024-11-03 13:48:36 +0900 [error]: /Users/watson/.rbenv/versions/3.1.6/lib/ruby/gems/3.1.0/gems/rr-3.1.1/lib/rr/injections/double_injection.rb:143:in `remove_method'
  \n  2024-11-03 13:48:36 +0900 [error]: /Users/watson/.rbenv/versions/3.1.6/lib/ruby/gems/3.1.0/gems/rr-3.1.1/lib/rr/injections/double_injection.rb:143:in `block in recv'
  \n  2024-11-03 13:48:36 +0900 [error]: /Users/watson/.rbenv/versions/3.1.6/lib/ruby/gems/3.1.0/gems/rr-3.1.1/lib/rr/injections/double_injection.rb:142:in `class_eval'
  \n  2024-11-03 13:48:36 +0900 [error]: /Users/watson/.rbenv/versions/3.1.6/lib/ruby/gems/3.1.0/gems/rr-3.1.1/lib/rr/injections/double_injection.rb:142:in `recv'
  \n  2024-11-03 13:48:36 +0900 [error]: /Users/watson/src/fluentd/lib/fluent/plugin/out_forward/ack_handler.rb:115:in `read_ack_from_sock'
  \n  2024-11-03 13:48:36 +0900 [error]: /Users/watson/src/fluentd/lib/fluent/plugin/out_forward/ack_handler.rb:71:in `block in collect_response'
  \n  2024-11-03 13:48:36 +0900 [error]: /Users/watson/src/fluentd/lib/fluent/plugin/out_forward/ack_handler.rb:70:in `each'
  \n  2024-11-03 13:48:36 +0900 [error]: /Users/watson/src/fluentd/lib/fluent/plugin/out_forward/ack_handler.rb:70:in `collect_response'
  \n  2024-11-03 13:48:36 +0900 [error]: test/plugin/out_forward/test_ack_handler.rb:124:in `block (3 levels) in <class:AckHandlerTest>'
  .
  <false> expected but was
  <true>
test/plugin/out_forward/test_ack_handler.rb:131:in `block in <class:AckHandlerTest>'
     128:       end
     129:
     130:       assert_true threads.map{ |t| t.join(10) }.all?
  => 131:       assert_false(
     132:         $log.out.logs.any? { |log| log.include?('[error]') },
     133:         $log.out.logs.select{ |log| log.include?('[error]') }.join('\n')
     134:       )
=============================================================================================================================================================================================================================
: (1.015446)

Finished in 1.015658 seconds.
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
1 tests, 2 assertions, 1 failures, 0 errors, 0 pendings, 0 omissions, 0 notifications
0% passed
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
0.98 tests/s, 1.97 assertions/s
rake aborted!
Command failed with status (1): [bundle exec ruby -w -I"lib:test" -Eascii-8bit:ascii-8bit test/plugin/out_forward/test_ack_handler.rb -v --name "test: raises no error when another thread closes a socket"]
/Users/watson/src/fluentd/runner.rake:4:in `block (2 levels) in <top (required)>'
/Users/watson/src/fluentd/runner.rake:2:in `times'
/Users/watson/src/fluentd/runner.rake:2:in `block in <top (required)>'
Tasks: TOP => default
(See full trace by running task with --trace)

Expected behavior

No errors

Your Environment

- Fluentd version:
- Package version:
- Operating system:
- Kernel version:

Your Configuration

none

Your Error Log

See "To Reproduce" section

Additional context

No response

[Watson1978]

It seems that the stubbed method is removed is unstable in rr gem.

https://github.com/rr/rr/blob/master/doc/03_api_overview.md#stub

#stub replaces the method on the object with only an implementation.

The stub is intended to replace the method.
However, the object returned by IO.pipe does not have recv method.

irb(main):001> r, w = IO.pipe
=> [#<IO:fd 6>, #<IO:fd 7>]
irb(main):002> r.methods.grep(/recv/)
=> []

If I add recv method into the object, it makes the test to stable.

diff --git a/test/plugin/out_forward/test_ack_handler.rb b/test/plugin/out_forward/test_ack_handler.rb
index 8264e4ac..8cf3cc31 100644
--- a/test/plugin/out_forward/test_ack_handler.rb
+++ b/test/plugin/out_forward/test_ack_handler.rb
@@ -111,6 +111,7 @@ class AckHandlerTest < Test::Unit::TestCase
     r, w = IO.pipe
     begin
       w.write(chunk_id)
+      def r.recv(*args) ; end
       stub(r).recv { |_|
         sleep(1) # To ensure that multiple threads select the socket before closing.
         raise IOError, 'stream closed in another thread' if r.closed?

The test was successful even after repeating it at least 500 times.

[Watson1978]

I think there is no need to use stub method.

diff --git a/test/plugin/out_forward/test_ack_handler.rb b/test/plugin/out_forward/test_ack_handler.rb
index 8264e4ac..9246da86 100644
--- a/test/plugin/out_forward/test_ack_handler.rb
+++ b/test/plugin/out_forward/test_ack_handler.rb
@@ -111,11 +111,11 @@ class AckHandlerTest < Test::Unit::TestCase
     r, w = IO.pipe
     begin
       w.write(chunk_id)
-      stub(r).recv { |_|
+      def r.recv(arg)
         sleep(1) # To ensure that multiple threads select the socket before closing.
-        raise IOError, 'stream closed in another thread' if r.closed?
+        raise IOError, 'stream closed in another thread' if self.closed?
         MessagePack.pack({ 'ack' => Base64.encode64('chunk_id 111') })
-      }
+      end
       ack.enqueue(r)
 
       threads = []

[daipom]

Thanks for investigating this!
It appears that the race condition of the stub method definition causes this failure.
Let's stop using stub for this test as #4694 (comment).