Fix random test hangs when using proxy module

When one end of the connection is closed for some reason, the proxy
should close the other end as well. Otherwise random hangs are possible
when a client doesn't notice that the server has closed the socket.

Previously clients never noticed leader disconnect (they only noticed
that the heartbeats were missing), and some tests relied on that
behaviour while proxy is paused. To preserve it, do not close the
sockets while proxy is paused.

Resolves #298

Author: sergepetrenko

luatest/replica_conn.lua

@@ -37,15 +37,15 @@ function Connection:initialize()
         self.process_client = {
             pre = nil,
             func = self.forward_to_server,
-            post = self.close_client_socket,
+            post = self.stop,
         }
     end
 
     if self.process_server == nil then
         self.process_server = {
             pre = nil,
             func = self.forward_to_client,
-            post = self.close_server_socket,
+            post = self.stop,
         }
     end
 
@@ -71,7 +71,7 @@ function Connection:process_socket(sock, process)
     local f = fiber.new(function()
         if process.pre ~= nil then process.pre(self) end
 
-        while sock:peer() do
+        while sock:peer() or not self.running do
             if not self.running then
                 fiber.sleep(TIMEOUT)
             elseif sock:readable(TIMEOUT) then

test/proxy_test.lua

@@ -1,6 +1,10 @@
 local t = require('luatest')
 local proxy = require('luatest.replica_proxy')
 local utils = require('luatest.utils')
+local replica_set = require('luatest.replica_set')
+local server = require('luatest.server')
+
+local fiber = require('fiber')
 
 local g = t.group('proxy-version-check')
 
@@ -14,3 +18,66 @@ g.test_proxy_errors = function()
                                     server_socket_path = 'somepath'
                                 })
 end
+
+local g1 = t.group('proxy', {
+    {is_paused = true},
+    {is_paused = false}
+})
+
+g1.before_all(function(cg)
+    -- Proxy only works on tarantool 2.10+
+    t.run_only_if(utils.version_ge(utils.get_tarantool_version(),
+                                   utils.version(2, 10, 1)),
+                  [[Proxy works on Tarantool 2.10.1+.
+                    See tarantool/tarantool@57ecb6cd90b4 for details]])
+    cg.rs = replica_set:new{}
+    cg.box_cfg = {
+        replication_timeout = 0.1,
+        replication = {
+            server.build_listen_uri('server2_proxy', cg.rs.id),
+        },
+    }
+    cg.server1 = cg.rs:build_and_add_server{
+        alias = 'server1',
+        box_cfg = cg.box_cfg,
+    }
+    cg.box_cfg.replication = nil
+    cg.server2 = cg.rs:build_and_add_server{
+        alias = 'server2',
+        box_cfg = cg.box_cfg,
+    }
+    cg.proxy = proxy:new{
+        client_socket_path = server.build_listen_uri('server2_proxy', cg.rs.id),
+        server_socket_path = server.build_listen_uri('server2', cg.rs.id),
+    }
+    t.assert(cg.proxy:start{force = true}, 'Proxy is started')
+    cg.rs:start{}
+end)
+
+g1.test_server_disconnect_is_noticed = function(cg)
+    local id = cg.server2:get_instance_id()
+    t.helpers.retrying({}, cg.server1.assert_follows_upstream, cg.server1, id)
+    if cg.params.is_paused then
+        cg.proxy:pause()
+    end
+    cg.server2:stop()
+    fiber.sleep(cg.box_cfg.replication_timeout)
+    local upstream = cg.server1:exec(function(upstream_id)
+        return box.info.replication[upstream_id].upstream
+    end, {id})
+    if cg.params.is_paused then
+        t.assert_equals(upstream.status, 'follow',
+                        'Server disconnect is not noticed')
+    else
+        t.assert_equals(upstream.system_message, 'Broken pipe',
+                        'Server disconnect is noticed')
+    end
+    if cg.params.is_paused then
+        cg.proxy:resume()
+    end
+    cg.server2:start()
+end
+
+g1.after_all(function(cg)
+    cg.rs:drop()
+end)