replicaset: reconnect after fiber kill

Currently if we kill the worker fiber of the connection, which was
initialized with 'reconnect_after' option, this connection goes into
'error_reconnect' or 'error' state (depends on tarantool version).
Reconnecting doesn't happen in both cases and the only way for user
to return router to working order is reloading or manual restoring of
the connections.

This patch introduces reconnecting in that case. It should be used
wisely, though. Fiber's killing doesn't happen instantly and if the
user doesn't wait util fiber's status is 'dead' and makes the request
immediately, exception will be probably thrown as the fiber can die
in the middle of request.

Closes #341

test/router-luatest/router_test.lua

@@ -545,3 +545,34 @@ g.test_enable_disable = function(g)
     -- we don't want this server to interfere with subsequent tests
     g.router_1:drop()
 end
+
+g.test_explicit_fiber_kill = function(g)
+    local bids = vtest.cluster_exec_each_master(g, function()
+        return _G.get_first_bucket()
+    end)
+
+    -- Kill worker fibers of connections to masters
+    g.router:exec(function()
+        for _, rs in ipairs(ivshard.router.static.replicasets) do
+            -- Explicitly kill the connection's worker fiber.
+            -- Callback for pushes is executed inside this fiber.
+            pcall(function()
+                rs.master.conn:eval([[
+                    box.session.push(nil)
+                ]], {}, {on_push = function()
+                    ifiber.self():cancel()
+                end})
+            end)
+        end
+    end)
+
+    -- Check that the replicaset is still accessible
+    g.router:exec(function(bids)
+        for _, bid in ipairs(bids) do
+            local res, err = ivshard.router.callrw(bid, 'echo', {1},
+                                                   {timeout = iwait_timeout})
+            ilt.assert(res, 1)
+            ilt.assert(err, nil)
+        end
+    end, {bids})
+end

vshard/replicaset.lua

@@ -144,6 +144,40 @@ local function netbox_wait_connected(conn, timeout)
     return timeout
 end
 
+--
+-- Check if the connection is dead and it won't be restored automatically.
+-- Even though replicaset's connections are initialized with the option
+-- 'reconnect_after', there's cases, when no reconnect will be done (e.g.
+-- if the conn was explicitly cancelled or the connection's worker fiber
+-- was killed). In these situations we need to reestablish it manually as
+-- the missing connection to some replicaset is a critical problem which
+-- leads to the inability of the user to access some part of a data via the
+-- router.
+--
+-- Note: the function expects the conn to be non-nil.
+--
+local function netbox_is_conn_dead(conn)
+    -- Fast path - conn is almost always 'active'.
+    if conn.state == 'active' then
+        return false
+    end
+    if conn.state == 'error' or conn.state == 'closed' then
+        return true
+    end
+    if conn.state ~= 'error_reconnect' then
+        -- The connection is fetching schema or doing auth, which
+        -- means it is not dead and shoudn't be reinitialized.
+        return false
+    end
+    if not conn._fiber then
+        -- The fiber field is not present in old netbox, which correctly
+        -- reports state as 'error' when its fiber got killed. It would
+        -- be filtered out above then.
+        return false
+    end
+    return conn._fiber:status() ~= "dead"
+end
+
 --
 -- Check if the replica is not in backoff. It also serves as an update - if the
 -- replica still has an old backoff timestamp, it is cleared. This way of
@@ -170,7 +204,7 @@ end
 --
 local function replicaset_connect_to_replica(replicaset, replica)
     local conn = replica.conn
-    if not conn or conn.state == 'closed' then
+    if not conn or netbox_is_conn_dead(conn) then
         conn = netbox.connect(replica.uri, {
             reconnect_after = consts.RECONNECT_TIMEOUT,
             wait_connected = false