[PATCH] lua/trigger: cleanup lua stack after trigger run

[Serge Petrenko <sergepetrenko@tarantool.org>]

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
---
https://github.com/tarantool/tarantool/issues/4275
https://github.com/tarantool/tarantool/tree/sp/gh-4275-before-replace-on-recovery

 src/lua/trigger.c                |  6 ++++
 test/box/before_replace.result   | 48 ++++++++++++++++++++++++++++++++
 test/box/before_replace.test.lua | 27 ++++++++++++++++++
 test/box/box.lua                 | 23 +++++++++++++++
 4 files changed, 104 insertions(+)

diff --git a/src/lua/trigger.c b/src/lua/trigger.c
index ec4d8aab3..4803e85c5 100644
--- a/src/lua/trigger.c
+++ b/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);
 }
 
diff --git a/test/box/before_replace.result b/test/box/before_replace.result
index 5054fdefd..661d999bb 100644
--- a/test/box/before_replace.result
+++ b/test/box/before_replace.result
@@ -858,3 +858,51 @@ save_type
 s:drop()
 ---
 ...
+--
+-- gh-4275 segfault if before_replace
+-- trigger set in on_schema_init triggers runs on recovery.
+--
+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,66 do gen_inserts() end
+---
+...
+test_run:cmd('restart server default with args="true"')
+s = box.space.test_on_schema_init
+---
+...
+s:count()
+---
+- 1
+...
+-- insertion: 65999 updates from gen_inserts(), starting value is 1.
+-- recovery: 66000 invocations of before replace trigger, each
+-- invocation updates the tuple, but only in memory: the WAL remains
+-- unchanged.
+-- A total of 1 + 65999 + 66000 = 132000
+s:get{1}[2] == 2 * 66000 or s:get{1}[2]
+---
+- true
+...
+test_run:cmd('restart server default with cleanup=1')
diff --git a/test/box/before_replace.test.lua b/test/box/before_replace.test.lua
index c46ec25a0..688b012d0 100644
--- a/test/box/before_replace.test.lua
+++ b/test/box/before_replace.test.lua
@@ -306,3 +306,30 @@ _ = s:insert{1}
 save_type
 
 s:drop()
+
+--
+-- gh-4275 segfault if before_replace
+-- trigger set in on_schema_init triggers runs on recovery.
+--
+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,66 do gen_inserts() end
+test_run:cmd('restart server default with args="true"')
+s = box.space.test_on_schema_init
+s:count()
+-- insertion: 65999 updates from gen_inserts(), starting value is 1.
+-- recovery: 66000 invocations of before replace trigger, each
+-- invocation updates the tuple, but only in memory: the WAL remains
+-- unchanged.
+-- A total of 1 + 65999 + 66000 = 132000
+s:get{1}[2] == 2 * 66000 or s:get{1}[2]
+test_run:cmd('restart server default with cleanup=1')
diff --git a/test/box/box.lua b/test/box/box.lua
index 2a8e0e4fa..7eaa5043e 100644
--- a/test/box/box.lua
+++ b/test/box/box.lua
@@ -1,5 +1,28 @@
 #!/usr/bin/env tarantool
 os = require('os')
+local enable_before_replace = arg[1] ~= nil
+
+print('\n\n\n', enable_before_replace, '\n\n\n')
+
+function test_before_replace_trig(old, new)
+    return 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()
+    if enable_before_replace then
+        box.space._space:on_replace(space_on_replace_trig)
+    end
+end
+
+box.ctl.on_schema_init(on_init_trig)
 
 box.cfg{
     listen              = os.getenv("LISTEN"),
-- 
2.20.1 (Apple Git-117)