Re-connect on fork(2)

.rubocop.yml

@@ -23,7 +23,7 @@ Style/Documentation:
     - 'test/**/*'
 
 Metrics/MethodLength:
-  Max: 20
+  Max: 25
 
 ThreadSafety:
   Enabled: true

lib/dalli/protocol/connection_manager.rb

@@ -100,13 +100,13 @@ def socket_timeout
 
       def confirm_ready!
         close if request_in_progress?
-        close_on_fork if fork_detected?
+        reconnect_on_fork if fork_detected?
       end
 
       def confirm_in_progress!
         raise '[Dalli] No request in progress. This may be a bug in Dalli.' unless request_in_progress?
 
-        close_on_fork if fork_detected?
+        reconnect_on_fork if fork_detected?
       end
 
       def close
@@ -237,13 +237,12 @@ def log_warn_message(err_or_string)
         end
       end
 
-      def close_on_fork
-        message = 'Fork detected, re-connecting child process...'
+      def reconnect_on_fork
+        message = 'Fork detected, re-connecting in child process...'
         Dalli.logger.info { message }
-        # Close socket on a fork, setting us up for reconnect
-        # on next request.
+        # Close socket on a fork and re-connect
         close
-        raise Dalli::RetryableNetworkError, message
+        establish_connection
       end
 
       def fork_detected?

lib/dalli/protocol/meta.rb

@@ -52,7 +52,6 @@ def write_multi_storage_req(_mode, pairs, ttl = nil, _cas = nil, options = {})
       end
 
       # rubocop:disable Metrics/AbcSize
-      # rubocop:disable Metrics/MethodLength
       # rubocop:disable Metrics/CyclomaticComplexity
       # rubocop:disable Metrics/PerceivedComplexity
       def read_multi_req(keys)
@@ -84,7 +83,6 @@ def read_multi_req(keys)
         results
       end
       # rubocop:enable Metrics/AbcSize
-      # rubocop:enable Metrics/MethodLength
       # rubocop:enable Metrics/CyclomaticComplexity
       # rubocop:enable Metrics/PerceivedComplexity
 

test/integration/test_concurrency.rb

@@ -10,9 +10,9 @@
 
       cache.set('f', 'zzz')
 
-      assert op_cas_succeeds((cache.cas('f') do |value|
+      assert op_cas_succeeds(cache.cas('f') do |value|
         value << 'z'
-      end))
+      end)
       assert_equal 'zzzz', cache.get('f')
 
       # Have a bunch of threads perform a bunch of operations at the same time.
@@ -59,9 +59,9 @@
 
       cache.set('f', 'zzz')
 
-      assert op_cas_succeeds((cache.cas('f') do |value|
+      assert op_cas_succeeds(cache.cas('f') do |value|
         value << 'z'
-      end))
+      end)
       assert_equal 'zzzz', cache.get('f')
 
       multi_keys = { 'ab' => 'vala', 'bb' => 'valb', 'cb' => 'valc' }
@@ -115,9 +115,9 @@
 
         cache.set('f', 'zzz')
 
-        assert op_cas_succeeds((cache.cas('f') do |value|
+        assert op_cas_succeeds(cache.cas('f') do |value|
           value << 'z'
-        end))
+        end)
         assert_equal 'zzzz', cache.get('f')
         multi_keys = { 'ab' => 'vala', 'bb' => 'valb', 'cb' => 'valc', 'dd' => 'vald' }
         cache.set_multi(multi_keys, 10)
@@ -170,9 +170,9 @@
 
       cache.set('f', 'zzz')
 
-      assert op_cas_succeeds((cache.cas('f') do |value|
+      assert op_cas_succeeds(cache.cas('f') do |value|
         value << 'z'
-      end))
+      end)
       assert_equal 'zzzz', cache.get('f')
 
       10.times do
@@ -216,9 +216,9 @@
 
       cache.set('f', 'zzz')
 
-      assert op_cas_succeeds((cache.cas('f') do |value|
+      assert op_cas_succeeds(cache.cas('f') do |value|
         value << 'z'
-      end))
+      end)
       assert_equal 'zzzz', cache.get('f')
 
       10.times do

test/integration/test_fork_safety.rb

@@ -0,0 +1,51 @@
+# frozen_string_literal: true
+
+require_relative '../helper'
+
+describe 'Fork safety' do
+  next unless Process.respond_to?(:fork)
+
+  it 'automatically reconnects after fork' do
+    memcached_persistent do |dc|
+      dc.set('fork_test_key', 'parent_value')
+
+      assert_equal 'parent_value', dc.get('fork_test_key')
+
+      # Fork a child process
+      read_pipe, write_pipe = IO.pipe
+      pid = fork do
+        read_pipe.close
+
+        # In the child process, we should detect the fork and reconnect
+        begin
+          dc.set('child_key', 'child_value')
+          value = dc.get('child_key')
+
+          write_pipe.write("success:#{value}")
+        rescue StandardError => e
+          write_pipe.write("error:#{e.class.name}:#{e.message}")
+        ensure
+          write_pipe.close
+          exit!(0)
+        end
+      end
+
+      # In the parent process
+      write_pipe.close
+
+      # Wait for child process to finish
+      Process.wait(pid)
+
+      # Read result from pipe
+      result = read_pipe.read
+      read_pipe.close
+
+      # Verify the child successfully reconnected and performed operations
+      assert_match(/^success:/, result, "Child process encountered an error: #{result}")
+      assert_equal 'success:child_value', result
+
+      # Parent should still be able to work
+      assert_equal 'parent_value', dc.get('fork_test_key')
+    end
+  end
+end

test/test_fork_safety.rb

@@ -0,0 +1,44 @@
+# frozen_string_literal: true
+
+require_relative 'helper'
+
+class TestForkSafety < Minitest::Spec
+  it 'remains operational after forking' do
+    skip unless Process.respond_to?(:fork)
+
+    memcached_persistent do |dc|
+      dc.set('key', 'foo')
+
+      assert_equal 'foo', dc.get('key')
+
+      pid = fork do
+        # Child process should detect fork and reconnect automatically
+        100.times do |i|
+          dc.set('key', "child_#{i}")
+          sleep(0.01)
+        end
+        exit!(0)
+      end
+
+      # Parent process should continue to work
+      100.times do |_i|
+        begin
+          dc.get('foo')
+        rescue StandardError
+          nil
+        end
+        sleep(0.01) # Add a small delay
+      end
+
+      # Wait for child to finish
+      _, status = Process.wait2(pid)
+
+      assert_predicate(status, :success?)
+
+      # Verify we can still perform operations in parent
+      dc.get('key')
+
+      assert_kind_of String, dc.get('key'), 'Expected a string value from memcached'
+    end
+  end
+end