lua/trigger: cleanup lua stack after trigger run

This patch adds a stack cleanup after a trigger is run and its return
values, if any, have been read.

This problem was found in a case when on_schema_init trigger set an
on_replace trigger on a space, and the trigger ran during recovery.
This lead to Lua stack overflows for the aforementioned reasons.

Closes #4275

src/lua/trigger.c

@@ -102,9 +102,15 @@ lbox_trigger_run(struct trigger *ptr, void *event)
 	int nret = lua_gettop(L) - top;
 	if (trigger->pop_event != NULL &&
 	    trigger->pop_event(L, nret, event) != 0) {
+		lua_settop(L, top);
 		luaL_unref(tarantool_L, LUA_REGISTRYINDEX, coro_ref);
 		diag_raise();
 	}
+	/*
+	 * Clear the stack after pop_event saves all
+	 * the needed return values.
+	 */
+	lua_settop(L, top);
 	luaL_unref(tarantool_L, LUA_REGISTRYINDEX, coro_ref);
 }
 

test/box/before_replace.result

@@ -858,3 +858,84 @@ save_type
 s:drop()
 ---
 ...
+--
+-- gh-4275 segfault if before_replace
+-- trigger set in on_schema_init triggers runs on recovery.
+--
+test_run:cmd('create server test with script="box/on_schema_init.lua"')
+---
+- true
+...
+test_run:cmd('start server test')
+---
+- true
+...
+test_run:cmd('switch test')
+---
+- true
+...
+s = box.schema.space.create('test_on_schema_init')
+---
+...
+_ = s:create_index('pk')
+---
+...
+test_run:cmd('setopt delimiter ";"')
+---
+- true
+...
+function gen_inserts()
+    box.begin()
+    for i = 1,1000 do
+        s:upsert({1, 1}, {{'+', 2, 1}})
+    end
+    box.commit()
+end;
+---
+...
+test_run:cmd('setopt delimiter ""');
+---
+- true
+...
+for i = 1,17 do gen_inserts() end
+---
+...
+test_run:cmd('restart server test')
+s = box.space.test_on_schema_init
+---
+...
+s:count()
+---
+- 1
+...
+-- For this test the number of invocations of the before_replace
+-- trigger during recovery multiplied by the amount of return values
+-- in before_replace trigger must be greater than 66000.
+-- A total of 1 + 16999 + 17000 + 1= 34001
+-- insertion: 16999 updates from gen_inserts(), starting value is 1,
+-- plus additional 17000 runs of before_replace trigger, each updating
+-- the value by adding 1.
+-- recovery: 17000 invocations of before replace trigger.
+-- Each invocation updates the recovered tuple, but only in-memory, the
+-- WAL remains intact. Thus, we only see the last update. During recovery
+-- the value is only increased by 1, and this change is visible only in memory.
+s:get{1}[2] == 1 + 16999 + 17000 + 1 or s:get{1}[2]
+---
+- true
+...
+test_run:cmd('switch default')
+---
+- true
+...
+test_run:cmd('stop server test')
+---
+- true
+...
+test_run:cmd('cleanup server test')
+---
+- true
+...
+test_run:cmd('delete server test')
+---
+- true
+...

test/box/before_replace.test.lua

@@ -306,3 +306,42 @@ _ = s:insert{1}
 save_type
 
 s:drop()
+
+--
+-- gh-4275 segfault if before_replace
+-- trigger set in on_schema_init triggers runs on recovery.
+--
+test_run:cmd('create server test with script="box/on_schema_init.lua"')
+test_run:cmd('start server test')
+test_run:cmd('switch test')
+s = box.schema.space.create('test_on_schema_init')
+_ = s:create_index('pk')
+test_run:cmd('setopt delimiter ";"')
+function gen_inserts()
+    box.begin()
+    for i = 1,1000 do
+        s:upsert({1, 1}, {{'+', 2, 1}})
+    end
+    box.commit()
+end;
+test_run:cmd('setopt delimiter ""');
+for i = 1,17 do gen_inserts() end
+test_run:cmd('restart server test')
+s = box.space.test_on_schema_init
+s:count()
+-- For this test the number of invocations of the before_replace
+-- trigger during recovery multiplied by the amount of return values
+-- in before_replace trigger must be greater than 66000.
+-- A total of 1 + 16999 + 17000 + 1= 34001
+-- insertion: 16999 updates from gen_inserts(), starting value is 1,
+-- plus additional 17000 runs of before_replace trigger, each updating
+-- the value by adding 1.
+-- recovery: 17000 invocations of before replace trigger.
+-- Each invocation updates the recovered tuple, but only in-memory, the
+-- WAL remains intact. Thus, we only see the last update. During recovery
+-- the value is only increased by 1, and this change is visible only in memory.
+s:get{1}[2] == 1 + 16999 + 17000 + 1 or s:get{1}[2]
+test_run:cmd('switch default')
+test_run:cmd('stop server test')
+test_run:cmd('cleanup server test')
+test_run:cmd('delete server test')

test/box/on_schema_init.lua

+#!/usr/bin/env tarantool
+os = require('os')
+
+function test_before_replace_trig(old, new)
+    -- return multiple values so that the stack fills earlier.
+    return new:update{{'+', 2, 1}}, new:update{{'+', 2, 1}}, new:update{{'+', 2, 1}}, new:update{{'+', 2, 1}}
+end
+
+function space_on_replace_trig(old, new)
+    if new and new[3] == 'test_on_schema_init' then
+        box.on_commit(function()
+            box.space.test_on_schema_init:before_replace(test_before_replace_trig)
+        end)
+    end
+end
+
+function on_init_trig()
+    box.space._space:on_replace(space_on_replace_trig)
+end
+
+box.ctl.on_schema_init(on_init_trig)
+
+box.cfg{
+    listen = os.getenv("LISTEN")
+}
+
+require('console').listen(os.getenv('ADMIN'))