net.box: reference the connection object from asynchronous requests

In order to prevent the garbage collection of the discarded connection,
asynchronous requests must reference the connection object. We must
reference the connection object rather than the transport object, because
our garbage collection hook is attached to the former.

Closes #9629

NO_DOC=<bugfix>

(cherry picked from commit fb5bf51)

changelogs/unreleased/gh-9629-netbox-async-request-connection-gc.md

@@ -0,0 +1,4 @@
+## bugfix/lua
+
+* Fixed a bug in `net.box` when a connection with asynchronous requests could
+  get garbage collected (gh-9629).

src/box/lua/net_box.c

@@ -350,6 +350,12 @@ struct netbox_request {
 	 * the response hasn't been received yet.
 	 */
 	struct error *error;
+	/**
+	 * Lua reference to the connection object if the request is asynchronous
+	 * or LUA_NOREF if the request is synchronous. Used to prevent garbage
+	 * collection in case the user discards the connection.
+	 */
+	int remote_ref;
 };
 
 /*
@@ -392,6 +398,7 @@ netbox_request_destroy(struct netbox_request *request)
 	luaL_unref(tarantool_L, LUA_REGISTRYINDEX, request->index_ref);
 	if (request->error != NULL)
 		error_unref(request->error);
+	luaL_unref(tarantool_L, LUA_REGISTRYINDEX, request->remote_ref);
 }
 
 /**
@@ -2513,6 +2520,7 @@ luaT_netbox_transport_make_request(struct lua_State *L, int idx,
 	request->index_ref = LUA_NOREF;
 	request->result_ref = LUA_NOREF;
 	request->error = NULL;
+	request->remote_ref = LUA_NOREF;
 	netbox_request_register(request, transport);
 	return 0;
 }
@@ -2521,9 +2529,13 @@ static int
 luaT_netbox_transport_perform_async_request(struct lua_State *L)
 {
 	struct netbox_transport *transport = luaT_check_netbox_transport(L, 1);
+	/* The connection object. */
+	assert(lua_istable(L, 2));
 	struct netbox_request *request = lua_newuserdata(L, sizeof(*request));
-	if (luaT_netbox_transport_make_request(L, 2, transport, request) != 0)
+	if (luaT_netbox_transport_make_request(L, 3, transport, request) != 0)
 		return luaT_push_nil_and_error(L);
+	lua_pushvalue(L, 2);
+	request->remote_ref = luaL_ref(L, LUA_REGISTRYINDEX);
 	luaL_getmetatable(L, netbox_request_typename);
 	lua_setmetatable(L, -2);
 	return 1;

src/box/lua/net_box.lua

@@ -890,7 +890,7 @@ function remote_methods:_request_impl(method, opts, format, stream_id, ...)
             if opts.on_push or opts.on_push_ctx then
                 error('To handle pushes in an async request use future:pairs()')
             end
-            return transport:perform_async_request(buffer, skip_header,
+            return transport:perform_async_request(self, buffer, skip_header,
                                                    return_raw, table.insert,
                                                    {}, format, stream_id,
                                                    method, ...)

test/box-luatest/gh_9629_netbox_async_request_connection_gc_test.lua

@@ -0,0 +1,26 @@
+local net_box = require('net.box')
+local server = require('luatest.server')
+local t = require('luatest')
+
+local g = t.group()
+
+g.before_all(function(cg)
+    cg.server = server:new()
+    cg.server:start()
+end)
+
+g.after_all(function(cg)
+    cg.server:drop()
+end)
+
+-- Tests that a connection with asynchronous requests does not get garbage
+-- collected.
+g.test_async_request_connection_gc = function(cg)
+    local c = net_box.connect(cg.server.net_box_uri)
+    local f = c:eval('return 777', {}, {is_async = true})
+    c = nil -- luacheck: no unused
+    collectgarbage()
+    local res, err = f:wait_result(10)
+    t.assert_equals(err, nil)
+    t.assert_equals(res[1], 777)
+end