box: improve check for dangerous select calls

Consider the following case a safe `select` call (i.e., do not issue a
warning):
* offset + limit <= 1000 (default value of offset is 0, default value of
limit is 4294967295);

Add new dangerous `select` call case:
* 'ALL', 'GE', 'GT', 'LE', 'LT' iterator even with key present.

Because of the additional logic, it was decided to refactor
`check_select_args` and call it after resolving options in
`check_select_opts`.

Closes tarantool#7129

@TarantoolBot document
Title: improve check for dangerous `select` calls

Calls with `offset + limit <= 100`
are now considered safe (i.e., a warning is not issued), e.g.:
box.space.s:select(nil, {offset = 1000})
box.space.s:select(nil, {limit = 1000})
box.space.s:select(nil, {offset = 500, limit = 500})

'ALL', 'GE', 'GT', 'LE', 'LT' iterators are now considered dangerous by
default even with key present, e.g.:
box.space.s:select({0})
box.space.s:select({0}, {iterator = 'GE'})

But not that these calls are still safe:
box.space.s:select({0}, {limit = 1000})
box.space.s:select({0}, {limit = 1000, iterator = 'GE'})
box.space.s:select({0}, {iterator = 'EQ'})

Author: CuriousGeorgiy

changelogs/unreleased/gh-7129-improve-check-for-dangerous-select-calls.md

@@ -0,0 +1,6 @@
+## feature/box
+
+* Improved check for dangerous select calls: calls with `offset + limit <= 100`
+  are now considered safe (i.e., a warning is not issued); 'ALL', 'GE', 'GT',
+  'LE', 'LT' iterators are now considered dangerous by default even with key
+  present (gh-7129).

src/box/lua/schema.lua

@@ -2352,28 +2352,34 @@ local function check_select_opts(opts, key_is_nil)
     local offset = 0
     local limit = 4294967295
     local iterator = check_iterator_type(opts, key_is_nil)
+    local fullscan = false
     if opts ~= nil and type(opts) == "table" then
         if opts.offset ~= nil then
             offset = opts.offset
         end
         if opts.limit ~= nil then
             limit = opts.limit
         end
+        if opts.fullscan ~= nil then
+            fullscan = opts.fullscan
+        end
     end
-    return iterator, offset, limit
+    return iterator, offset, limit, fullscan
 end
 
 box.internal.check_select_opts = check_select_opts -- for net.box
 
-local check_select_args_rl = log.internal.ratelimit:new()
-local function check_select_args(index, key, opts)
-    local rl = check_select_args_rl
-
-    if index.space_id >= 512 and
-       (type(key) == 'nil' or (type(key) == 'table' and next(key) == nil)) and
-       (opts == nil or not opts.fullscan) then
-        rl:log_crit('empty or nil `select` call on user space with id=%d\n %s',
-                    index.space_id, debug.traceback())
+local check_select_safety_rl = log.internal.ratelimit.new()
+local function check_select_safety(index, key_is_nil, itype, limit, offset,
+                                   fullscan)
+    local rl = check_select_safety_rl
+    local sid = index.space_id
+    local point_iter = itype == box.index.EQ or itype == box.index.REQ
+    local window = offset + limit
+    if sid >= 512 and (key_is_nil or not point_iter) and
+       (not fullscan and window > 1000) then
+        rl:log_crit("Potentially long select from space '%s' (%d)\n %s",
+                    box.space[sid].name, sid, debug.traceback())
     end
 end
 
@@ -2382,10 +2388,12 @@ base_index_mt.select_ffi = function(index, key, opts)
         return index:select_luac(key, opts)
     end
     check_index_arg(index, 'select')
-    check_select_args(index, key, opts)
     local ibuf = cord_ibuf_take()
     local key, key_end = tuple_encode(ibuf, key)
-    local iterator, offset, limit = check_select_opts(opts, key + 1 >= key_end)
+    local key_is_nil = key + 1 >= key_end
+    local iterator, offset, limit, fullscan =
+        check_select_opts(opts, key_is_nil)
+    check_select_safety(index, key_is_nil, iterator, limit, offset, fullscan)
 
     local nok = builtin.box_select(index.space_id, index.id, iterator, offset,
                                    limit, key, key_end, port) ~= 0
@@ -2406,9 +2414,11 @@ end
 
 base_index_mt.select_luac = function(index, key, opts)
     check_index_arg(index, 'select')
-    check_select_args(index, key, opts)
     local key = keify(key)
-    local iterator, offset, limit = check_select_opts(opts, #key == 0)
+    local key_is_nil = #key == 0
+    local iterator, offset, limit, fullscan =
+        check_select_opts(opts, key_is_nil)
+    check_select_safety(index, key_is_nil, iterator, limit, offset, fullscan)
     return internal.select(index.space_id, index.id, iterator,
         offset, limit, key)
 end

src/lua/log.lua

@@ -396,6 +396,16 @@ local function log_pid()
     return tonumber(ffi.C.log_pid)
 end
 
+local ratelimit_enabled = true
+
+local function ratelimit_enable()
+    ratelimit_enabled = true
+end
+
+local function ratelimit_disable()
+    ratelimit_enabled = false
+end
+
 local Ratelimit = {
     interval = 60,
     burst = 10,
@@ -404,6 +414,10 @@ local Ratelimit = {
     start = 0,
 }
 
+local function ratelimit_new(object)
+    return Ratelimit:new(object)
+end
+
 function Ratelimit:new(object)
     object = object or {}
     setmetatable(object, self)
@@ -412,8 +426,11 @@ function Ratelimit:new(object)
 end
 
 function Ratelimit:check()
-    local clock = require('clock')
+    if not ratelimit_enabled then
+        return 0, true
+    end
 
+    local clock = require('clock')
     local now = clock.monotonic()
     local saved_suppressed = 0
     if now > self.start + self.interval then
@@ -650,7 +667,11 @@ local log = {
         cfg_set_log_format = box_api_set_log_format,
     },
     internal = {
-        ratelimit = Ratelimit
+        ratelimit = {
+            new = ratelimit_new,
+            enable = ratelimit_enable,
+            disable = ratelimit_disable,
+        },
     }
 }
 

test/box-luatest/gh_6539_log_user_space_empty_or_nil_select_test.lua

@@ -7,24 +7,41 @@ local g = t.group()
 g.before_all(function()
     g.server = server:new{alias = 'dflt'}
     g.server:start()
+    g.server:exec(function()
+        require("log").internal.ratelimit.disable()
+    end)
 end)
 
 g.after_all(function()
     g.server:drop()
 end)
 
-local expected_log_entry = 'C> empty or nil `select` call on user space ' ..
-                           'with id=%d+'
+local expected_log_entry_fmt =
+    "C> Potentially long select from space '%s' %%(%d%%)"
 local grep_log_bytes = 256
-local dangerous_call_fmts = {'box.space.%s:select()', 'box.space.%s:select{}',
-                             'box.space.%s:select(nil, {})',
-                             'box.space.%s:select(nil, {fullscan = false})'}
+local dangerous_call_fmts = {
+    'box.space.%s:select(nil)',
+    'box.space.%s:select({})',
+    'box.space.%s:select(nil, {})',
+    'box.space.%s:select(nil, {iterator = "EQ"})',
+    'box.space.%s:select(nil, {iterator = "REQ"})',
+    'box.space.%s:select(nil, {fullscan = false})',
+    'box.space.%s:select(nil, {limit = 1001})',
+    'box.space.%s:select(nil, {offset = 1})',
+    'box.space.%s:select(nil, {limit = 500, offset = 501})',
+    'box.space.%s:select({0}, {limit = 1001, iterator = "ALL"})',
+    'box.space.%s:select({0}, {limit = 1001, iterator = "GE"})',
+    'box.space.%s:select({0}, {limit = 1001, iterator = "GT"})',
+    'box.space.%s:select({0}, {limit = 1001, iterator = "LE"})',
+    'box.space.%s:select({0}, {limit = 1001, iterator = "LT"})',
+}
 local log_entry_absence_msg_fmt = 'log must not contain a critical entry ' ..
                                   'about `%s` call on a %s %s space'
 
 g.test_log_entry_absence_for_sys_space = function()
-    t.fail_if(g.server:eval('return box.space._space.id') > 512,
-              '`_space` space is assumed to be a system space')
+    local sid = g.server:eval('return box.space._space.id')
+    t.fail_if(sid > 512, "'_space' space is assumed to be a system space")
+    local expected_log_entry = expected_log_entry_fmt:format("_space", sid)
     for _, call_fmt in pairs(dangerous_call_fmts) do
         local call = call_fmt:format('_space')
         g.server:eval(call)
@@ -40,21 +57,55 @@ for _, eng in pairs{'memtx', 'vinyl'} do
                        "{engine = '%s'})"):format(space, eng))
         g.server:eval(("box.space.%s:create_index('pk')"):format(space))
 
-        local safe_call_fmts = {'box.space.%s:select{0}',
-                                'box.space.%s:select({0}, {fullscan = false})',
-                                'box.space.%s:select({0}, {fullscan = true})'}
+        local sid = g.server:eval(('return box.space.%s.id'):format(space))
+        local expected_log_entry = expected_log_entry_fmt:format(space, sid)
+        require('log').info("expected_log_entry=%s", expected_log_entry)
+
+        local safe_call_fmts = {
+            'box.space.%s:select(nil, {limit = 1000, iter = "ALL"})',
+            'box.space.%s:select(nil, {limit = 1000, iter = "GE"})',
+            'box.space.%s:select(nil, {limit = 1000, iter = "GT"})',
+            'box.space.%s:select(nil, {limit = 1000, iter = "LE"})',
+            'box.space.%s:select(nil, {limit = 1000, iter = "LT"})',
+            'box.space.%s:select(nil, {limit = 500, iter = "EQ"})',
+            'box.space.%s:select(nil, {limit = 500, offset = 500})',
+            'box.space.%s:select(nil, {limit = 250, offset = 250})',
+            'box.space.%s:select({0}, {iter = "EQ"})',
+            'box.space.%s:select({0}, {iter = "REQ"})',
+            'box.space.%s:select({0}, {limit = 1001, iter = "EQ"})',
+            'box.space.%s:select({0}, {iter = "EQ", fullscan = false})',
+            'box.space.%s:select({0}, {iter = "ALL", fullscan = true})',
+            'box.space.%s:select({0}, {iter = "GE", fullscan = true})',
+            'box.space.%s:select({0}, {iter = "GT", fullscan = true})',
+            'box.space.%s:select({0}, {iter = "LE", fullscan = true})',
+            'box.space.%s:select({0}, {iter = "LT", fullscan = true})',
+            'box.space.%s:select({0}, {limit = 1001, fullscan = true})',
+        }
         for _, call_fmt in pairs(safe_call_fmts) do
             local call = call_fmt:format(space)
             g.server:eval(call)
             t.assert_not(g.server:grep_log(expected_log_entry, grep_log_bytes),
                          log_entry_absence_msg_fmt:format(call, eng, "user"))
         end
 
-        local explicit_call_fmts = {'box.space.%s:select(nil, ' ..
-                                    '{fullscan = true})',
-                                    'box.space.%s:select({}, ' ..
-                                    '{fullscan = true})'}
-        for _, call_fmt in pairs(explicit_call_fmts) do
+        local explicit_fullscan_call_fmts = {
+            'box.space.%s:select(nil, {fullscan = true})',
+            'box.space.%s:select(nil, {limit = 1001, fullscan = true})',
+            'box.space.%s:select(nil, {offset = 1001, fullscan = true})',
+            'box.space.%s:select(nil, {limit = 500, offset = 501, ' ..
+            'fullscan = true})',
+            'box.space.%s:select({0}, {limit = 1001, iterator = "ALL", ' ..
+            'fullscan = true})',
+            'box.space.%s:select({0}, {limit = 1001, iterator = "GE", ' ..
+            'fullscan = true})',
+            'box.space.%s:select({0}, {limit = 1001, iterator = "GT", ' ..
+            'fullscan = true})',
+            'box.space.%s:select({0}, {limit = 1001, iterator = "LE", ' ..
+            'fullscan = true})',
+            'box.space.%s:select({0}, {limit = 1001, iterator = "LT", ' ..
+            'fullscan = true})',
+        }
+        for _, call_fmt in pairs(explicit_fullscan_call_fmts) do
             local call = call_fmt:format(space)
             g.server:eval(call)
             t.assert_not(g.server:grep_log(expected_log_entry, grep_log_bytes),