box: make box.schema DDL functions transactional

Wraps multistatement DDL functions into begin/commit block if no
transaction is active.

The functions are:
- `box.schema.space.drop`
- `box.schema.index.create`
- `box.schema.index.drop`
- `box.schema.sequence.drop`
- `box.schema.func.drop`
- `box.schema.user.create`
- `box.schema.user.drop`
- `box.schema.role.drop`

Added tests for atomicity of each transactioned function except
the `box.schema.role.drop`, which is implicitly tested with the
`box.schema.user.drop` test, and the `box.schema.index.drop`,
which is impossible to test in this flavor whthout using error
injection.

Updated the tests modified in #8947, because
the space drop is atomic now.

TBD: Make `box.schema.index.alter` atomic.

Closes #4348

NO_DOC=bugfix

changelogs/unreleased/gh-4348-transactional-ddl.md

@@ -0,0 +1,4 @@
+## bugfix/box
+
+* All DDL functions from the `box.schema` module are now wrapped into a
+  transaction to avoid database inconsistency on failed operations (gh-4348).

src/box/lua/schema.lua

@@ -382,6 +382,18 @@ box.atomic = function(arg0, arg1, ...)
     end
 end
 
+-- Wrap a function into transaction if none is active.
+local function atomic_wrapper(func)
+    return function(...)
+        -- No reason to start a transaction of one is active already.
+        if box.is_in_txn() then
+            return func(...)
+        else
+            return box.atomic(func, ...)
+        end
+    end
+end
+
 -- box.commit yields, so it's defined as Lua/C binding
 -- box.rollback and box.rollback_to_savepoint yields as well
 
@@ -820,7 +832,7 @@ end
 
 box.schema.create_space = box.schema.space.create
 
-box.schema.space.drop = function(space_id, space_name, opts)
+box.schema.space.drop = atomic_wrapper(function(space_id, space_name, opts)
     check_param(space_id, 'space_id', 'number')
     opts = opts or {}
     check_param_table(opts, { if_exists = 'boolean' })
@@ -872,7 +884,7 @@ box.schema.space.drop = function(space_id, space_name, opts)
     end
 
     feedback_save_event('drop_space')
-end
+end)
 
 box.schema.space.rename = function(space_id, space_name)
     check_param(space_id, 'space_id', 'number')
@@ -1476,7 +1488,7 @@ local function func_id_by_name(func_name)
 end
 box.internal.func_id_by_name = func_id_by_name -- for space.upgrade
 
-box.schema.index.create = function(space_id, name, options)
+box.schema.index.create = atomic_wrapper(function(space_id, name, options)
     check_param(space_id, 'space_id', 'number')
     check_param(name, 'name', 'string')
     check_param_table(options, create_index_template)
@@ -1599,9 +1611,9 @@ box.schema.index.create = function(space_id, name, options)
 
     feedback_save_event('create_index')
     return space.index[name]
-end
+end)
 
-box.schema.index.drop = function(space_id, index_id)
+box.schema.index.drop = atomic_wrapper(function(space_id, index_id)
     check_param(space_id, 'space_id', 'number')
     check_param(index_id, 'index_id', 'number')
     if index_id == 0 then
@@ -1625,7 +1637,7 @@ box.schema.index.drop = function(space_id, index_id)
     _index:delete{space_id, index_id}
 
     feedback_save_event('drop_index')
-end
+end)
 
 box.schema.index.rename = function(space_id, index_id, name)
     check_param(space_id, 'space_id', 'number')
@@ -2850,7 +2862,7 @@ box.schema.sequence.alter = function(name, opts)
                       opts.max, opts.start, opts.cache, opts.cycle}
 end
 
-box.schema.sequence.drop = function(name, opts)
+box.schema.sequence.drop = atomic_wrapper(function(name, opts)
     opts = opts or {}
     check_param_table(opts, {if_exists = 'boolean'})
     local id = sequence_resolve(name)
@@ -2865,7 +2877,7 @@ box.schema.sequence.drop = function(name, opts)
     local _sequence_data = box.space[box.schema.SEQUENCE_DATA_ID]
     _sequence_data:delete{id}
     _sequence:delete{id}
-end
+end)
 
 local function privilege_parse(privs)
     -- TODO: introduce a global privilege -> bit mapping?
@@ -3142,7 +3154,7 @@ box.schema.func.create = function(name, opts)
                          opts.comment, opts.created, opts.last_altered}
 end
 
-box.schema.func.drop = function(name, opts)
+box.schema.func.drop = atomic_wrapper(function(name, opts)
     opts = opts or {}
     check_param_table(opts, { if_exists = 'boolean' })
     local _func = box.space[box.schema.FUNC_ID]
@@ -3165,7 +3177,7 @@ box.schema.func.drop = function(name, opts)
     end
     revoke_object_privs('function', fid)
     _func:delete{fid}
-end
+end)
 
 function box.schema.func.exists(name_or_id)
     local _vfunc = box.space[box.schema.VFUNC_ID]
@@ -3326,7 +3338,7 @@ box.schema.user.passwd = function(name, new_password)
     end
 end
 
-box.schema.user.create = function(name, opts)
+box.schema.user.create = atomic_wrapper(function(name, opts)
     local uid = user_or_role_resolve(name)
     opts = opts or {}
     check_param_table(opts, { password = 'string', if_not_exists = 'boolean' })
@@ -3356,7 +3368,7 @@ box.schema.user.create = function(name, opts)
     -- grant option
     box.session.su('admin', box.schema.user.grant, uid, 'session,usage', 'universe',
                    nil, {if_not_exists=true})
-end
+end)
 
 box.schema.user.exists = function(name)
     if user_resolve(name) then
@@ -3542,7 +3554,7 @@ box.schema.user.disable = function(user)
                             {if_exists = true})
 end
 
-box.schema.user.drop = function(name, opts)
+box.schema.user.drop = atomic_wrapper(function(name, opts)
     opts = opts or {}
     check_param_table(opts, { if_exists = 'boolean' })
     local uid = user_resolve(name)
@@ -3563,7 +3575,7 @@ box.schema.user.drop = function(name, opts)
         box.error(box.error.NO_SUCH_USER, name)
     end
     return
-end
+end)
 
 local function info(id)
     local _priv = box.space._vpriv
@@ -3616,7 +3628,7 @@ box.schema.role.create = function(name, opts)
                          math.floor(fiber.time())}
 end
 
-box.schema.role.drop = function(name, opts)
+box.schema.role.drop = atomic_wrapper(function(name, opts)
     opts = opts or {}
     check_param_table(opts, { if_exists = 'boolean' })
     local uid = role_resolve(name)
@@ -3632,7 +3644,7 @@ box.schema.role.drop = function(name, opts)
         box.error(box.error.DROP_USER, name, "the user or the role is a system")
     end
     return drop(uid)
-end
+end)
 
 local function role_check_grant_revoke_of_sys_priv(priv)
     priv = string.lower(priv)

test/box-luatest/builtin_events_test.lua

@@ -278,7 +278,7 @@ g.test_box_schema = function(cg)
 
     version_n = 0
     cg.master:exec(function() box.space.p:drop() end)
-    t.helpers.retrying({}, function() t.assert_equals(version_n, 2) end)
+    t.helpers.retrying({}, function() t.assert_equals(version_n, 1) end)
     -- there'll be 2 changes - index and space
     t.assert_equals(version, init_version + 4)