25461733485

Committed 06 May 2026 09:15PM UTC coverage: 88.457% (+0.1%) from 88.349%

Build # 25461733485

Build Type

push

github

Committed by

p0rtale

Commit Message

fix: handle nil bucket_id in storage get/update/delete operations

Starting from crud 1.7.0, storage asserts if `opts.bucket_id`
is nil during `get`, `update`, or `delete` operations.
Since older routers (< 1.7.0) do not pass `bucket_id`, this
causes protocol incompatibility.

This commit adds a nil check to skip bucket ref when `bucket_id`
is missing, restoring backward compatibility with old routers.
It also introduces a rate-limited warning and the
`tnt_crud_storage_nil_bucket_id_compat_total` metric to track
this state.

Coverage Stats

59 of 66 new or added lines in 6 files covered. (89.39%)

2 existing lines in 1 file now uncovered.

5303 of 5995 relevant lines covered (88.46%)

12326.38 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

92.73

/crud/delete.lua

local checks = require('checks')
local errors = require('errors')

local call = require('crud.common.call')
local compat_warn = require('crud.common.compat_warn')
local const = require('crud.common.const')
local utils = require('crud.common.utils')
local sharding = require('crud.common.sharding')
local sharding_key_module = require('crud.common.sharding.sharding_key')
local sharding_metadata_module = require('crud.common.sharding.sharding_metadata')
local dev_checks = require('crud.common.dev_checks')
local schema = require('crud.common.schema')
local bucket_ref_unref = require('crud.common.sharding.bucket_ref_unref')

local DeleteError = errors.new_class('DeleteError', {capture_stack = false})

local delete = {}

local DELETE_FUNC_NAME = 'delete_on_storage'
local CRUD_DELETE_FUNC_NAME = utils.get_storage_call(DELETE_FUNC_NAME)

local function delete_on_storage(space_name, key, field_names, opts)
    dev_checks('string', '?', '?table', {
        bucket_id = '?number|cdata',
        sharding_key_hash = '?number',
        sharding_func_hash = '?number',
        skip_sharding_hash_check = '?boolean',
        noreturn = '?boolean',
        fetch_latest_metadata = '?boolean',
    })

    opts = opts or {}

    local space = box.space[space_name]
    if space == nil then
        return nil, DeleteError:new("Space %q doesn't exist", space_name)
    end

    local _, err = sharding.check_sharding_hash(space_name,
                                                opts.sharding_func_hash,
                                                opts.sharding_key_hash,
                                                opts.skip_sharding_hash_check)

    if err ~= nil then
        return nil, err
    end

    -- Skip bucket reference if bucket_id is not provided to support old routers.
    local unref = nil
    if opts.bucket_id ~= nil then
        local ref_ok, bucket_ref_err, unref_func = bucket_ref_unref.bucket_refrw(opts.bucket_id, space.engine)
        if not ref_ok then
            return nil, bucket_ref_err
        end
        unref = unref_func
    else
        compat_warn.log_nil_bucket_id('delete', space_name, space.engine)
    end

    -- add_space_schema_hash is false because
    -- reloading space format on router can't avoid delete error on storage
    local result =  schema.wrap_func_result(space, space.delete, {
        add_space_schema_hash = false,
        field_names = field_names,
        noreturn = opts.noreturn,
        fetch_latest_metadata = opts.fetch_latest_metadata,
    }, space, key)

    if unref ~= nil then
        local unref_ok, err_unref = unref(opts.bucket_id, space.engine)
        if not unref_ok then
            return nil, err_unref
        end
    end

    return result
end

delete.storage_api = {[DELETE_FUNC_NAME] = delete_on_storage}

-- returns result, err, need_reload
-- need_reload indicates if reloading schema could help
-- see crud.common.schema.wrap_func_reload()
local function call_delete_on_router(vshard_router, space_name, key, opts)
    dev_checks('table', 'string', '?', {
        timeout = '?number',
        bucket_id = '?',
        fields = '?table',
        vshard_router = '?string|table',
        noreturn = '?boolean',
        fetch_latest_metadata = '?boolean',
    })

    local space, err, netbox_schema_version = utils.get_space(space_name, vshard_router, opts.timeout)
    if err ~= nil then
        return nil, DeleteError:new("An error occurred during the operation: %s", err), const.NEED_SCHEMA_RELOAD
    end
    if space == nil then
        return nil, DeleteError:new("Space %q doesn't exist", space_name), const.NEED_SCHEMA_RELOAD
    end

    if box.tuple.is(key) then
        key = key:totable()
    end

    local sharding_key_hash = nil
    local skip_sharding_hash_check = nil

    local sharding_key = key
    if opts.bucket_id == nil then
        if space.index[0] == nil then
            return nil, DeleteError:new("Cannot fetch primary index parts"), const.NEED_SCHEMA_RELOAD
        end
        local primary_index_parts = space.index[0].parts

        local sharding_key_data, err = sharding_metadata_module.fetch_sharding_key_on_router(vshard_router, space_name)
        if err ~= nil then
            return nil, err
        end

        sharding_key, err = sharding_key_module.extract_from_pk(vshard_router,
                                                                space_name,
                                                                sharding_key_data.value,
                                                                primary_index_parts, key)
        if err ~= nil then
            return nil, err
        end

        sharding_key_hash = sharding_key_data.hash
    else
        skip_sharding_hash_check = true
    end

    local bucket_id_data, err = sharding.key_get_bucket_id(vshard_router, space_name, sharding_key, opts.bucket_id)
    if err ~= nil then
        return nil, err
    end

    -- When the sharding index (bucket_id) is the primary index, bucket_id can be passed as box.NULL.
    sharding.fill_bucket_id_pk(space, key, bucket_id_data.bucket_id)

    local delete_on_storage_opts = {
        bucket_id = bucket_id_data.bucket_id,
        sharding_func_hash = bucket_id_data.sharding_func_hash,
        sharding_key_hash = sharding_key_hash,
        skip_sharding_hash_check = skip_sharding_hash_check,
        noreturn = opts.noreturn,
        fetch_latest_metadata = opts.fetch_latest_metadata,
    }

    local call_opts = {
        mode = 'write',
        timeout = opts.timeout,
    }

    local storage_result, err = call.single(vshard_router,
        bucket_id_data.bucket_id, CRUD_DELETE_FUNC_NAME,
        {space_name, key, opts.fields, delete_on_storage_opts},
        call_opts
    )

    if err ~= nil then
        local err_wrapped = DeleteError:new("Failed to call delete on storage-side: %s", err)

        if sharding.result_needs_sharding_reload(err) then
            return nil, err_wrapped, const.NEED_SHARDING_RELOAD
        end

        return nil, err_wrapped
    end

    if storage_result.err ~= nil then
        return nil, DeleteError:new("Failed to delete: %s", storage_result.err)
    end

    if opts.noreturn == true then
        return nil
    end

    local tuple = storage_result.res

    if opts.fetch_latest_metadata == true then
        -- This option is temporary and is related to [1], [2].
        -- [1] https://github.com/tarantool/crud/issues/236
        -- [2] https://github.com/tarantool/crud/issues/361
        space = utils.fetch_latest_metadata_when_single_storage(space, space_name, netbox_schema_version,
                                                                vshard_router, opts, storage_result.storage_info)
    end

    return utils.format_result({tuple}, space, opts.fields)
end

--- Deletes tuple from the specified space by key
--
-- @function call
--
-- @param string space_name
--  A space name
--
-- @param key
--  Primary key value
--
-- @tparam ?number opts.timeout
--  Function call timeout
--
-- @tparam ?number opts.bucket_id
--  Bucket ID
--  (by default, it's vshard.router.bucket_id_strcrc32 of primary key)
--
-- @tparam ?string|table opts.vshard_router
--  Cartridge vshard group name or vshard router instance.
--  Set this parameter if your space is not a part of the
--  default vshard cluster.
--
-- @tparam ?boolean opts.noreturn
--  Suppress returning successfully processed tuple.
--
-- @return[1] object
-- @treturn[2] nil
-- @treturn[2] table Error description
--
function delete.call(space_name, key, opts)
    checks('string', '?', {
        timeout = '?number',
        bucket_id = '?',
        fields = '?table',
        vshard_router = '?string|table',
        noreturn = '?boolean',
        fetch_latest_metadata = '?boolean',
    })

    opts = opts or {}

    local vshard_router, err = utils.get_vshard_router_instance(opts.vshard_router)
    if err ~= nil then
        return nil, DeleteError:new(err)
    end

    return schema.wrap_func_reload(vshard_router, sharding.wrap_method, call_delete_on_router,
                                   space_name, key, opts)
end

return delete

1	local checks = require('checks')	633✔
2	local errors = require('errors')	633✔
3
4	local call = require('crud.common.call')	633✔
5	local compat_warn = require('crud.common.compat_warn')	633✔
6	local const = require('crud.common.const')	633✔
7	local utils = require('crud.common.utils')	633✔
8	local sharding = require('crud.common.sharding')	633✔
9	local sharding_key_module = require('crud.common.sharding.sharding_key')	633✔
10	local sharding_metadata_module = require('crud.common.sharding.sharding_metadata')	633✔
11	local dev_checks = require('crud.common.dev_checks')	633✔
12	local schema = require('crud.common.schema')	633✔
13	local bucket_ref_unref = require('crud.common.sharding.bucket_ref_unref')	633✔
14
15	local DeleteError = errors.new_class('DeleteError', {capture_stack = false})	633✔
16
17	local delete = {}	633✔
18
19	local DELETE_FUNC_NAME = 'delete_on_storage'	633✔
20	local CRUD_DELETE_FUNC_NAME = utils.get_storage_call(DELETE_FUNC_NAME)	633✔
21
22	local function delete_on_storage(space_name, key, field_names, opts)
23	dev_checks('string', '?', '?table', {	154✔
24	bucket_id = '?number\|cdata',
25	sharding_key_hash = '?number',
26	sharding_func_hash = '?number',
27	skip_sharding_hash_check = '?boolean',
28	noreturn = '?boolean',
29	fetch_latest_metadata = '?boolean',
30	})
31
32	opts = opts or {}	154✔
33
34	local space = box.space[space_name]	154✔
35	if space == nil then	154✔
36	return nil, DeleteError:new("Space %q doesn't exist", space_name)	×
37	end
38
39	local _, err = sharding.check_sharding_hash(space_name,	308✔
40	opts.sharding_func_hash,	154✔
41	opts.sharding_key_hash,	154✔
42	opts.skip_sharding_hash_check)	154✔
43
44	if err ~= nil then	154✔
45	return nil, err	4✔
46	end
47
48	-- Skip bucket reference if bucket_id is not provided to support old routers.
49	local unref = nil	150✔
50	if opts.bucket_id ~= nil then	150✔
51	local ref_ok, bucket_ref_err, unref_func = bucket_ref_unref.bucket_refrw(opts.bucket_id, space.engine)	146✔
52	if not ref_ok then	146✔
NEW 53	return nil, bucket_ref_err	×
54	end
55	unref = unref_func	146✔
56	else
57	compat_warn.log_nil_bucket_id('delete', space_name, space.engine)	4✔
58	end
59
60	-- add_space_schema_hash is false because
61	-- reloading space format on router can't avoid delete error on storage
62	local result = schema.wrap_func_result(space, space.delete, {	300✔
63	add_space_schema_hash = false,
64	field_names = field_names,	150✔
65	noreturn = opts.noreturn,	150✔
66	fetch_latest_metadata = opts.fetch_latest_metadata,	150✔
67	}, space, key)	150✔
68
69	if unref ~= nil then	150✔
70	local unref_ok, err_unref = unref(opts.bucket_id, space.engine)	146✔
71	if not unref_ok then	146✔
NEW 72	return nil, err_unref	×
73	end
74	end
75
76	return result	150✔
77	end
78
79	delete.storage_api = {[DELETE_FUNC_NAME] = delete_on_storage}	633✔
80
81	-- returns result, err, need_reload
82	-- need_reload indicates if reloading schema could help
83	-- see crud.common.schema.wrap_func_reload()
84	local function call_delete_on_router(vshard_router, space_name, key, opts)
85	dev_checks('table', 'string', '?', {	170✔
86	timeout = '?number',
87	bucket_id = '?',
88	fields = '?table',
89	vshard_router = '?string\|table',
90	noreturn = '?boolean',
91	fetch_latest_metadata = '?boolean',
92	})
93
94	local space, err, netbox_schema_version = utils.get_space(space_name, vshard_router, opts.timeout)	170✔
95	if err ~= nil then	170✔
96	return nil, DeleteError:new("An error occurred during the operation: %s", err), const.NEED_SCHEMA_RELOAD	×
97	end
98	if space == nil then	170✔
99	return nil, DeleteError:new("Space %q doesn't exist", space_name), const.NEED_SCHEMA_RELOAD	32✔
100	end
101
102	if box.tuple.is(key) then	308✔
103	key = key:totable()	×
104	end
105
106	local sharding_key_hash = nil	154✔
107	local skip_sharding_hash_check = nil
108
109	local sharding_key = key	154✔
110	if opts.bucket_id == nil then	154✔
111	if space.index[0] == nil then	142✔
112	return nil, DeleteError:new("Cannot fetch primary index parts"), const.NEED_SCHEMA_RELOAD	×
113	end
114	local primary_index_parts = space.index[0].parts	142✔
115
116	local sharding_key_data, err = sharding_metadata_module.fetch_sharding_key_on_router(vshard_router, space_name)	142✔
117	if err ~= nil then	142✔
118	return nil, err	×
119	end
120
121	sharding_key, err = sharding_key_module.extract_from_pk(vshard_router,	284✔
122	space_name,	142✔
123	sharding_key_data.value,	142✔
124	primary_index_parts, key)	284✔
125	if err ~= nil then	142✔
126	return nil, err	4✔
127	end
128
129	sharding_key_hash = sharding_key_data.hash	138✔
130	else
131	skip_sharding_hash_check = true	12✔
132	end
133
134	local bucket_id_data, err = sharding.key_get_bucket_id(vshard_router, space_name, sharding_key, opts.bucket_id)	150✔
135	if err ~= nil then	150✔
136	return nil, err	2✔
137	end
138
139	-- When the sharding index (bucket_id) is the primary index, bucket_id can be passed as box.NULL.
140	sharding.fill_bucket_id_pk(space, key, bucket_id_data.bucket_id)	148✔
141
142	local delete_on_storage_opts = {	148✔
143	bucket_id = bucket_id_data.bucket_id,	148✔
144	sharding_func_hash = bucket_id_data.sharding_func_hash,	148✔
145	sharding_key_hash = sharding_key_hash,	148✔
146	skip_sharding_hash_check = skip_sharding_hash_check,	148✔
147	noreturn = opts.noreturn,	148✔
148	fetch_latest_metadata = opts.fetch_latest_metadata,	148✔
149	}
150
151	local call_opts = {	148✔
152	mode = 'write',
153	timeout = opts.timeout,	148✔
154	}
155
156	local storage_result, err = call.single(vshard_router,	296✔
157	bucket_id_data.bucket_id, CRUD_DELETE_FUNC_NAME,	148✔
158	{space_name, key, opts.fields, delete_on_storage_opts},	148✔
159	call_opts
160	)	148✔
161
162	if err ~= nil then	148✔
163	local err_wrapped = DeleteError:new("Failed to call delete on storage-side: %s", err)	2✔
164
165	if sharding.result_needs_sharding_reload(err) then	4✔
166	return nil, err_wrapped, const.NEED_SHARDING_RELOAD	2✔
167	end
168
169	return nil, err_wrapped	×
170	end
171
172	if storage_result.err ~= nil then	146✔
173	return nil, DeleteError:new("Failed to delete: %s", storage_result.err)	68✔
174	end
175
176	if opts.noreturn == true then	112✔
177	return nil	2✔
178	end
179
180	local tuple = storage_result.res	110✔
181
182	if opts.fetch_latest_metadata == true then	110✔
183	-- This option is temporary and is related to [1], [2].
184	-- [1] https://github.com/tarantool/crud/issues/236
185	-- [2] https://github.com/tarantool/crud/issues/361
186	space = utils.fetch_latest_metadata_when_single_storage(space, space_name, netbox_schema_version,	4✔
187	vshard_router, opts, storage_result.storage_info)	4✔
188	end
189
190	return utils.format_result({tuple}, space, opts.fields)	110✔
191	end
192
193	--- Deletes tuple from the specified space by key
194	--
195	-- @function call
196	--
197	-- @param string space_name
198	-- A space name
199	--
200	-- @param key
201	-- Primary key value
202	--
203	-- @tparam ?number opts.timeout
204	-- Function call timeout
205	--
206	-- @tparam ?number opts.bucket_id
207	-- Bucket ID
208	-- (by default, it's vshard.router.bucket_id_strcrc32 of primary key)
209	--
210	-- @tparam ?string\|table opts.vshard_router
211	-- Cartridge vshard group name or vshard router instance.
212	-- Set this parameter if your space is not a part of the
213	-- default vshard cluster.
214	--
215	-- @tparam ?boolean opts.noreturn
216	-- Suppress returning successfully processed tuple.
217	--
218	-- @return[1] object
219	-- @treturn[2] nil
220	-- @treturn[2] table Error description
221	--
222	function delete.call(space_name, key, opts)	633✔
223	checks('string', '?', {	162✔
224	timeout = '?number',
225	bucket_id = '?',
226	fields = '?table',
227	vshard_router = '?string\|table',
228	noreturn = '?boolean',
229	fetch_latest_metadata = '?boolean',
230	})
231
232	opts = opts or {}	162✔
233
234	local vshard_router, err = utils.get_vshard_router_instance(opts.vshard_router)	162✔
235	if err ~= nil then	162✔
236	return nil, DeleteError:new(err)	8✔
237	end
238
239	return schema.wrap_func_reload(vshard_router, sharding.wrap_method, call_delete_on_router,	158✔
240	space_name, key, opts)	158✔
241	end
242
243	return delete	633✔

tarantool / crud / 25461733485

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous