box: revert the lua space object on the space drop rollback

Before this commit the space rollback had been treated as a new space
creation, so it caused creation of a new space object in the Lua's
box.space namespace. Since the preceding space drop removed the space
object from the namespace, on the space rollback all the Lua users of
the space loosed the track of its changes: the original space object
is never updated anymore.

This is fixed by detecting the space rollback and restoring the old
space object instead of creating a new one.

Closes #9120

NO_DOC=bugfix

changelogs/unreleased/gh-9120-space-drop-rollback.md

+## bugfix/box
+
+* Fixed a bug when local references to a space object got out of sync with the
+  space on the space drop rollback (gh-9120).

src/box/lua/space.cc

@@ -615,6 +615,26 @@ box_lua_space_new(struct lua_State *L, struct space *space)
 		lua_setfield(L, -2, "space");
 		lua_getfield(L, -1, "space");
 	}
+	/*
+	 * We can have the following conditions here:
+	 * a) The space is totally new (e. g. on new space creation).
+	 * b) The space is replaced (e. g. on index update).
+	 * c) The space is updated (e. g. on sequence update).
+	 * d) The space is reverted (e. g. on space drop rollback).
+	 *
+	 * - In case a) we need to create new box.space.<id|name> entries.
+	 * - In cases b) and c) we need to update the existing box.space.<id>
+	 *   entry, drop the old box.space.<name> and create a new one.
+	 * - In case d) we need to restore the original box.space.<id|name>
+	 *   entries, but let's only restore the box.space.<id> entry and
+	 *   perform the same actions as for b) and c) - it won't change the
+	 *   visible behavior but will make the code simpler.
+	 */
+	if (space->lua_ref != LUA_NOREF) {
+		/* We have either case c) or d). */
+		lua_rawgeti(L, LUA_REGISTRYINDEX, space->lua_ref);
+		lua_rawseti(L, -2, space_id(space));
+	}
 	lua_rawgeti(L, -1, space_id(space));
 	if (lua_isnil(L, -1)) {
 		/*
@@ -636,6 +656,18 @@ box_lua_space_new(struct lua_State *L, struct space *space)
 	lbox_fillspace(L, space, lua_gettop(L));
 	lua_setfield(L, -2, space_name(space));
 
+	if (space->lua_ref == LUA_NOREF) {
+		/*
+		 * Save the reference to the box.space[id] to restore
+		 * the exactly same object on the space drop rollback.
+		 * It prevents the situations when old references to
+		 * the space go out of sync with the space which had
+		 * been rolled back. For more information see #9120.
+		 */
+		lua_rawgeti(L, -1, space_id(space));
+		space->lua_ref = luaL_ref(L, LUA_REGISTRYINDEX);
+	}
+
 	lua_pop(L, 2); /* box, space */
 }
 

src/box/space.c

@@ -56,6 +56,7 @@
 #include "wal_ext.h"
 #include "coll_id_cache.h"
 #include "func_adapter.h"
+#include "lua/utils.h"
 
 int
 access_check_space(struct space *space, user_access_t access)
@@ -536,6 +537,7 @@ space_create(struct space *space, struct engine *engine,
 	space->constraint_ids = mh_strnptr_new();
 	rlist_create(&space->memtx_stories);
 	rlist_create(&space->alter_stmts);
+	space->lua_ref = LUA_NOREF;
 	return 0;
 
 fail_free_indexes:
@@ -642,6 +644,7 @@ space_delete(struct space *space)
 	mh_strnptr_delete(space->constraint_ids);
 	assert(space->sql_triggers == NULL);
 	assert(rlist_empty(&space->space_cache_pin_list));
+	luaL_unref(tarantool_L, LUA_REGISTRYINDEX, space->lua_ref);
 	space->vtab->destroy(space);
 }
 

src/box/space.h

@@ -280,6 +280,12 @@ struct space {
 	 * Linked by `coll_id_cache_holder::in_space`.
 	 */
 	struct rlist coll_id_holders;
+	/**
+	 * A reference to the lua table representing this space. Only used
+	 * on space drop rollback in order to keep the lua references to
+	 * this object in sync. For more information see #9120.
+	 */
+	int lua_ref;
 };
 
 /** Space alter statement. */

test/box-luatest/gh_9120_space_drop_rollback_test.lua

+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)
+
+g.after_each(function(cg)
+    cg.server:exec(function()
+        if box.space.test ~= nil then
+            box.space.test:drop()
+        end
+        if box.space.renamed ~= nil then
+            box.space.renamed:drop()
+        end
+    end)
+end)
+
+-- Checks that the space object of a dropped and rolled back space
+-- is kept in sync with the space from the box.space namespace.
+g.test_drop_rollback = function(cg)
+    cg.server:exec(function()
+        local s = box.schema.space.create('test')
+
+        -- Drop the space and roll it back.
+        box.begin()
+        box.space.test:drop()
+        box.rollback()
+
+        -- The local variable and the entry from the box.space
+        -- namespace reference the same object.
+        t.assert(s == box.space.test)
+    end)
+end
+
+-- The same check but with data inserts and rename involved.
+g.test_rename_drop_rollback = function(cg)
+    cg.server:exec(function()
+        local s = box.schema.space.create('test')
+
+        s:create_index('pk')
+        s:insert({1, 1})
+
+        -- Rename and drop the space and roll it back.
+        -- Do some inserts in the way.
+        box.begin()
+        box.space.test:insert({2, 2})
+        box.space.test:rename('renamed')
+        box.space.renamed:insert({3, 3})
+        box.space.renamed:drop()
+        box.rollback()
+
+        -- The local variable and the entry from the box.space
+        -- namespace reference the same object.
+        t.assert(s == box.space.test)
+
+        -- The space still has the same name.
+        t.assert_equals(s.name, 'test')
+
+        -- Contents hadn't changed.
+        t.assert_equals(s:select(), {{1, 1}})
+    end)
+end
+
+-- The same check but with more complex transaction with an attempt to
+-- trick a system by creating another space with the same ID and name.
+g.test_sophisticated_transaction = function(cg)
+    cg.server:exec(function()
+        local s = box.schema.space.create('test', {format = {'id'}})
+
+        s:create_index('pk')
+        s:insert({1, 1})
+
+        -- More sophisticated transaction: drop the old space, create a new one
+        -- with the same name but with a different format, roll everything back.
+        box.begin()
+        box.space.test:insert({2, 2})
+        box.space.test:drop()
+        box.schema.space.create('test', {format = {'identifier'}})
+        box.space.test:create_index('pk')
+        box.space.test:insert({3, 3})
+        box.rollback()
+
+        -- The local variable and the entry from the
+        -- box.space namespace are the same objects.
+        t.assert(s == box.space.test)
+
+        -- The space format and contents still the same.
+        t.assert_equals(s:format(), {{name = 'id', type = 'any'}})
+        t.assert_equals(s:select(), {{1, 1}})
+    end)
+end

test/box/errinj.result

@@ -418,11 +418,6 @@ s_withdata:drop()
 ---
 - error: Failed to write to disk
 ...
--- FIXME(gh-9120): The space is recreated, so the s_withdata object is not
--- valid anymore.
-s_withdata = box.space.withdata
----
-...
 box.space['withdata'].enabled
 ---
 - true
@@ -1343,11 +1338,6 @@ s:drop()
 ---
 - error: Failed to write to disk
 ...
--- FIXME(gh-9120): The space object is invalidated on reverted drop, so we set
--- it to the current one.
-s = box.space.test
----
-...
 s:truncate()
 ---
 - error: Failed to write to disk

test/box/errinj.test.lua

@@ -100,9 +100,6 @@ s_withindex.index.secondary
 s_withdata.index.secondary:drop()
 s_withdata.index.secondary.unique
 s_withdata:drop()
--- FIXME(gh-9120): The space is recreated, so the s_withdata object is not
--- valid anymore.
-s_withdata = box.space.withdata
 box.space['withdata'].enabled
 index4 = s_withdata:create_index('another', { type = 'tree', parts = { 5, 'unsigned' }, unique = false})
 s_withdata.index.another
@@ -450,9 +447,6 @@ errinj.set('ERRINJ_WAL_IO', true)
 s:drop()
 s:truncate()
 s:drop()
--- FIXME(gh-9120): The space object is invalidated on reverted drop, so we set
--- it to the current one.
-s = box.space.test
 s:truncate()
 errinj.set('ERRINJ_WAL_IO', false)
 for i = 1, 10 do s:replace{i + 10} end