xqueue.lua

local M = {}

local ffi = require 'ffi'
local log = require 'log'
local fiber = require 'fiber'
local clock = require 'clock'

local tuple_ctype = ffi.typeof(box.tuple.new())

local monotonic_max_age = 10*365*86400;

local function table_clear(t)
	if type(t) ~= 'table' then
		error("bad argument #1 to 'clear' (table expected, got "..(t ~= nil and type(t) or 'no value')..")",2)
	end
	local count = #t
	for i=0, count do t[i]=nil end
	return
end

local function is_array(t)
	local gen,param,state = ipairs(t)
	-- print(gen,param,state)
	local v = gen(param,state)
	-- print(v)
	return v ~= nil
end

local peers = {}

rawset(_G,"\0xq.on_connect",box.session.on_connect(function()
	local sid = box.session.id()
	local peer = box.session.peer()
	box.session.storage.peer = box.session.peer()
	peers[ sid ] = peer
	log.debug("connected %s, sid=%s, fid=%s", peer, sid, fiber.id() )
end,rawget(_G,"\0xq.on_connect")))

rawset(_G,"\0xq.on_disconnect",box.session.on_disconnect(function()
	local sid = box.session.id()
	local peer = peers[ sid ]
	peers[ sid ] = nil
	log.debug("disconnected %s, sid=%s, fid=%s", peer, sid, fiber.id() )
end,rawget(_G,"\0xq.on_disconnect")))


--[[

Field sets:
1. id, status (minimal required configuration)
2. id, status, priority
3. id, status, runat
4. id, status, priority, runat

Primary index variants:
1. index(status, [ id ])
2. index(status, priority, [ id ])


Format:
local format = box.space._space.index.name:get(space_name)[ 7 ]

Status:
	R - ready - task is ready to be taken
		created by put without delay
		turned on by release/kick without delay
		turned on from W when delay passed
		deleted after ttl if ttl is enabled

	T - taken - task is taken by consumer. may not be taken by other.
		turned into R after ttr if ttr is enabled

	W - waiting - task is not ready to be taken. waiting for its `delay`
		requires `runat`
		`delay` may be set during put, release, kick
		turned into R after delay

	B - buried - task was temporary discarded from queue by consumer
		may be revived using kick by administrator
		use it in unpredicted conditions, when man intervention is required
		Without mandatory monitoring (stats.buried) usage of buried is useless and awry

	Z - zombie - task was processed and ack'ed and *temporary* kept for delay

	D - done - task was processed and ack'ed and permanently left in database
		enabled when keep feature is set

	X - reserved for statistics

(TODO: reload/upgrade and feature switch)

Interface:

# Creator methods:

	M.upgrade(space, {
		format = {
			-- space format. applied to space.format() if passed
		},
		fields = {
			-- id is always taken from pk
			status   = 'status_field_name'    | status_field_no,
			runat    = 'runat_field_name'     | runat_field_no,
			priority = 'priority_field_name'  | priority_field_no,
		},
		features = {
			id = 'auto_increment' | 'uuid' | 'required' | function
				-- auto_increment - if pk is number, then use it for auto_increment
				-- uuid - if pk is string, then use uuid for id
				-- required - primary key MUST be present in tuple during put
				-- function - funciton will be called to aquire id for task

			buried = true,           -- if true, support bury/kick
			delayed = true,          -- if true, support delayed tasks, requires `runat`

			keep = true,             -- if true, keep ack'ed tasks in [D]one state, instead of deleting
				-- mutually exclusive with zombie

			zombie = true|number,    -- requires `runat` field
				-- if number, then with default zombie delay, otherwise only if set delay during ack
				-- mutually exclusive with keep

			ttl    = true|number,    -- requires `runat` field
				-- if number, then with default ttl, otherwise only if set during put/release
			ttr    = true|number,    -- requires `runat` field
				-- if number, then with default ttl, otherwise only if set
		},
	})

Producer methods:
	sp:put({...}, [ attr ]) (array or table (if have format) or tuple)

Consumer methods:

* id:
	- array of keyfields of pk
	- table of named keyields
	- tuple

sp:take(timeout) -> tuple
sp:ack(id, [ attr ])
	attr.update (only if support zombie)
sp:release(id, [ attr ])
	attr.ttr (only if support ttr)
	attr.ttl (only if support ttl)
	attr.update
sp:bury(id, [ attr ])
	attr.ttl (only if support ttl)
	attr.update

Admin methods:

sp:queue_stats()
sp:kick(N | id, [attr]) -- put buried task id or N oldest buried tasks to [R]eady


]]

local json = require 'json'
json.cfg{ encode_invalid_as_nil = true }


local function typeeq(src, ref)
	if ref == 'str' then
		return src == 'STR' or src == 'str' or src == 'string'
	elseif ref == 'num' then
		return src == 'NUM' or src == 'num' or src == 'number' or src == 'unsigned'
	else
		return src == ref
	end
end

local function is_eq(a, b, depth)
	local t = type(a)
	if t ~= type(b) then return false end
	if t == 'table' then
		for k,v in pairs(a) do
			if b[k] == nil then return end
			if not is_eq(v,b[k]) then return false end
		end
		for k,v in pairs(b) do
			if a[k] == nil then return end
			if not is_eq(v,a[k]) then return false end
		end
		return true
	elseif t == 'string' or t == 'number' then
		return a == b
	elseif t == 'cdata' then
		return ffi.typeof(a) == ffi.typeof(b) and a == b
	else
		error("Wrong types for equality", depth or 2)
	end
end


local function _tuple2table ( qformat )
	local rows = {}
	for k,v in ipairs(qformat) do
		table.insert(rows,"\t['"..v.name.."'] = t["..tostring(k).."];\n")
	end
	local fun = "return function(t,...) "..
		"if select('#',...) > 0 then error('excess args',2) end "..
		"return t and {\n"..table.concat(rows, "").."} or nil end\n"
	return dostring(fun)
end

local function _table2tuple ( qformat )
	local rows = {}
	for _,v in ipairs(qformat) do
		table.insert(rows,"\tt['"..v.name.."'] == nil and NULL or t['"..v.name.."'];\n")
	end
	local fun = "local NULL = require'msgpack'.NULL return function(t) return "..
		"t and box.tuple.new({\n"..table.concat(rows, "").."}) or nil end\n"
	-- print(fun)
	return dostring(fun)
end

local pretty_st = {
	R = "Ready",
	T = "Taken",
	W = "Waiting",
	B = "Buried",
	Z = "Zombie",
	D = "Done",
}

---@class xqueue.space
local methods = {}

---@class PrimaryKeyField:table
---@field no number position in tuple
---@field name string name of the field
---@field type "uuid"|"string"|"number"|"unsigned"|"integer"|"boolean" type of the field

---@class xqFeatures: table
---@field id "auto_increment"|"time64"|"uuid"| fun(): scalar (Default: uuid)
---@field retval "tuple" | "table"
---@field buried boolean
---@field delayed boolean
---@field keep boolean
---@field tube boolean
---@field zombie boolean|number
---@field zombie_delay number
---@field ttl boolean|number
---@field ttr boolean|number
---@field ttl_default number?
---@field ttr_default number?

---@class xq:table
---@field NEVER integer (Default: 0)
---@field atomic fun(self: xq, key: scalar, fun: fun(...:any): ...?): ...
---@field bysid table<number,table<string,string>> mapping sid => {key => key}
---@field taken table<string,number> mapping key => sid
---@field _lock table<string,boolean> locks key => boolean (in atomic)
---@field put_wait table<string,{cond:fiber.cond,task:box.tuple?,processed: boolean?}> mapping key => fiber.cond for producer
---@field take_wait fiber.channel
---@field take_chans table<string,fiber.channel> mapping tube => fiber.channel
---@field debug boolean
---@field have_runat boolean
---@field gen_id? fun(): scalar
---@field getkey fun(self: xq, arg: table|scalar|box.tuple): scalar
---@field packkey fun(self: xq, key: any): string
---@field tube_index? boxIndex
---@field index boxIndex
---@field key PrimaryKeyField
---@field fieldmap table<string,number>
---@field timeoffset fun(delta: number): number
---@field features xqFeatures
---@field fields table<string, string|number>
---@field tuple fun(tbl: table): box.tuple
---@field table fun(tuple: box.tuple): table
---@field retwrap fun(t: box.tuple|table): table|box.tuple
---@field wakeup fun(self: xq, t: box.tuple|table)
---@field runat_chan fiber.channel
---@field check_owner fun(self: xq, key: tuple_type|box.tuple): box.tuple
---@field put_back fun(key: table|box.tuple)
---@field _stat { counts: table, transition: table }
---@field putback fun(self: xq, task: table|box.tuple)
---@field _default_truncate fun(space: boxSpaceObject)
---@field _on_repl replaceTrigger
---@field _on_dis fun()
---@field ready? fiber.channel channel appears when xq is not ready


---@class xqueue.space: boxSpaceObject
---@field xq xq xqueue specific storage

---@class xqueue.fields
---@field status? string|number Xqueue name for the Status field
---@field runat? string|number Xqueue name for the RunAt field
---@field priority? string|number Xqueue name for Task priority field
---@field tube? string|number Xqueue name for Tube field

---@class xqueue.features
---@field id 'uuid' | 'auto_increment' | 'required' | 'time64' | (fun(): number|string) Mandatory Field for TaskID generation
---@field retval? 'table'|'tuple' Type of return Value of the task in xqueue methods.
---(Default: table when space with format, tuple when space is without format)
---@field buried? boolean should xqueue allow buring of the tasks (by default: true)
---@field keep? boolean should xqueue keep :ack'ed tasks in the Space or not
---@field delayed? boolean should xqueue allow delay tasks (requires runat field and index) (defauled: false)
---@field zombie? boolean|number should xqueue temporarily keep :ack'ed tasks in the Space (default: false). Mutually exclusive with keep
---when zombie is configured with number, then this value is treated as zombie_delay (requires runat to be present)
---@field ttl? boolean|number should xqueue allow Time-To-Live on tasks. When specified with number, this value used for ttl_default.
---Requires runat field and index.
---@field ttr? boolean|number should xqueue allow Time-To-Release on tasks. When specified with number, this value used for ttl_default.
---Requires runat field and index.

---@class xqueue.upgrade.options
---@field format? boxSpaceFormat
---@field fields xqueue.fields
---@field debug? boolean
---@field tube_stats? string[] List of tube names for per-tube statistics
---@field features xqueue.features
---@field worker? fun(task: box.tuple|table) simple ad-hoc worker callback
---@field workers? number (number of workers to spawn)

---Upgrades given space to xqueue instance
---@param space xqueue.space
---@param opts xqueue.upgrade.options
---@param depth? number
function M.upgrade(space,opts,depth)
	depth = depth or 0
	log.info("xqueue upgrade(%s,%s)", space.name, json.encode(opts))
	if not opts.fields then error("opts.fields required",2) end
	if opts.format then
		-- todo: check if already have such format
		local format_av = box.space._space.index.name:get(space.name)[ 7 ]
		if not is_eq(format_av, opts.format) then
			space:format(opts.format)
		else
			print("formats are equal")
		end
	end

	-- This variable will be defined later
	local taken_mt

	local self = {}
	if space.xq then
		self.taken = space.xq.taken
		self._stat = space.xq._stat
		self.bysid = space.xq.bysid
		self._lock = space.xq._lock
		self.take_wait = space.xq.take_wait
		self.take_chans = space.xq.take_chans or setmetatable({}, { __mode = 'v' })
		self.put_wait = space.xq.put_wait or setmetatable({}, { __mode = 'v' })
		self._on_repl  = space.xq._on_repl
		self._on_dis = space.xq._on_dis
	else
		self.taken = {}
		self.bysid = {}
		-- byfid = {};
		self._lock = {}
		self.put_wait = setmetatable({}, { __mode = 'v' })
		self.take_wait = fiber.channel(0)
		self.take_chans = setmetatable({}, { __mode = 'v' })
	end
	setmetatable(self.bysid, {
		__serialize='map',
		__newindex = function(t, key, val)
			if type(val) == 'table' then
				rawset(t, key, setmetatable(val, taken_mt))
			else
				rawset(t, key, val)
			end
		end
	})
	self.debug = not not opts.debug

	if not self._default_truncate then
		self._default_truncate = space.truncate
	end

	local format_av = box.space._space.index.name:get(space.name)[ 7 ]
	local format = {}
	local have_format = false
	local have_runat = false
	for no,f in pairs(format_av) do
		format[ f.name ] = {
			name = f.name;
			type = f.type;
			no   = no;
		}
		format[ no ] = format[ f.name ];
		have_format = true
		self.have_format = true
	end
	for _,idx in pairs(space.index) do
		for _,part in pairs(idx.parts) do
			format[ part.fieldno ] = format[ part.fieldno ] or { no = part.fieldno }
			format[ part.fieldno ].type = part.type
		end
	end

	-- dd(format)

	-- 1. fields check
	local fields = {}
	local fieldmap = {}
	for _,f in pairs({{"status","str"},{"runat","num"},{"priority","num"},{"tube","str"}}) do
		local fname,ftype = f[1], f[2]
		local num = opts.fields[fname]
		if num then
			if type(num) == 'string' then
				if format[num] then
					fields[fname] = format[num].no;
				else
					error(string.format("unknown field %s for %s", num, fname),2 + depth)
				end
			elseif type(num) == 'number' then
				if format[num] then
					fields[fname] = num
				else
					error(string.format("unknown field %s for %s", num, fname),2 + depth)
				end
			else
				error(string.format("wrong type %s for field %s, number or string required",type(num),fname),2 + depth)
			end
			-- check type
			if format[num] then
				if not typeeq(format[num].type, ftype) then
					error(string.format("type mismatch for field %s, required %s, got %s",fname, ftype, format[fname].type),2+depth)
				end
			end
			fieldmap[fname] = format[num].name
		end
	end

	-- dd(fields)

	-- 2. index check

	local pk = space.index[0]
	if #pk.parts ~= 1 then
		error("Composite primary keys are not supported yet")
	end
	local pktype
	if typeeq( format[pk.parts[1].fieldno].type, 'str' ) then
		pktype = 'string'
		taken_mt = { __serialize = 'map' }
	elseif typeeq( format[pk.parts[1].fieldno].type, 'num' ) then
		pktype = 'number'
		taken_mt = {
			__serialize = 'map',
			__newindex = function(t, key, val)
				return rawset(t, tostring(ffi.cast("uint64_t", key)), val)
			end,
			__index = function(t, key)
				return rawget(t, tostring(ffi.cast("uint64_t", key)))
			end
		}
	else
		error("Unknown key type "..format[pk.parts[1].fieldno].type)
	end
	setmetatable(self.taken, taken_mt)

	local pkf = {
		no   = pk.parts[1].fieldno;
		name = format[pk.parts[1].fieldno].name;
		['type'] = pktype;
	}
	self.key = pkf
	self.fields = fields
	self.fieldmap = fieldmap


	function self:getkey(arg)
		local _type = type(arg)
		if _type == 'table' then
			if is_array(arg) then
				return arg[ self.key.no ]
			else
				return arg[ self.key.name ]
				-- pass
			end
		elseif _type == 'cdata' and ffi.typeof(arg) == tuple_ctype then
			return arg[ self.key.no ]
		elseif _type == self.key.type then
			return arg
		else
			error("Wrong key/task argument. Expected table or tuple or key", 3)
		end
	end

	function self.packkey(_, key)
		if type(key) == 'cdata' then
			return tostring(ffi.cast("uint64_t", key))
		else
			return key
		end
	end

	do
		local filter
		if fields.priority then
			filter = function(index)
				return #index.parts >= 3
					and index.parts[1].fieldno == fields.status
					and index.parts[2].fieldno == fields.priority
					and index.parts[3].fieldno == self.key.no
			end
		else
			filter = function(index)
				return #index.parts >= 2
					and index.parts[1].fieldno == fields.status
					and index.parts[2].fieldno == self.key.no
			end
		end
		for i,index in pairs(space.index) do
			if type(i) == 'number' and filter(index) then
				self.index = index
				break
			end
		end
		if not self.index then
			if fields.priority then
				error("not found index by status + priority + id",2+depth)
			else
				error("not found index by status + id",2+depth)
			end
		end

		if fields.tube then
			for n,index in pairs(space.index) do
				if type(n) == 'number' and index.parts[1].fieldno == fields.tube then
					local not_match = false
					for i = 2, #index.parts do
						if index.parts[i].fieldno ~= self.index.parts[i-1].fieldno then
							not_match = true
							break
						end
					end
					if not not_match then
						self.tube_index = index
						break
					end
				end
			end
			if not self.tube_index then
				if fields.priority then
					error("not found index by tube + status + priority + id",2+depth)
				else
					error("not found index by tube + status + id",2+depth)
				end
			end
		end
	end

	---@type boxIndex
	local runat_index
	if fields.runat then
		for _,index in pairs(space.index) do
			if type(_) == 'number' then
				if index.parts[1].fieldno == fields.runat then
					-- print("found",index.name)
					runat_index = index
					break
				end
			end
		end
		if not runat_index then
			error(string.format("fields.runat requires tree index with this first field in it"),2+depth)
		else
			local t = runat_index:pairs({0},{iterator = box.index.GT}):nth(1)
			if t then
				if t[ self.fields.runat ] < monotonic_max_age then
					error("!!! Queue contains monotonic runat. Consider updating tasks (https://github.com/moonlibs/xqueue/issues/2)")
				end
			end
			have_runat = true
		end
	end
	self.have_runat = have_runat

	---@type table<string, { counts: {}, transition: {} }>
	local stat_tube = {}
	if self.fields.tube and type(opts.tube_stats) == 'table' then
		for _, tube_name in ipairs(opts.tube_stats) do
			if type(tube_name) == 'string' then
				stat_tube[tube_name] = {
					counts = {},
					transition = {},
				}
			end
		end
	end

	if not self._stat then
		self._stat = {
			counts = {};
			transition = {};
			tube = stat_tube;
		}
		if self.fields.tube then
			for status in pairs(pretty_st) do
				self._stat.counts[status] = 0LL+self.index:count(status)
			end
			for tube in pairs(stat_tube) do
				for status in pairs(pretty_st) do
					stat_tube[tube].counts[status] = 0LL+self.tube_index:count({ tube, status })
				end
			end
		else
			for status in pairs(pretty_st) do
				self._stat.counts[status] = 0LL+self.index:count(status)
			end
		end
	else
		self._stat = { counts = {}, transition = {}, tube = stat_tube }
	end

	-- 3. features check

	local features = {}

	local gen_id
	opts.features = opts.features or {}
	if not opts.features.id then
		opts.features.id = 'uuid'
	end
	-- 'auto_increment' | 'time64' | 'uuid' | 'required' | function
	if opts.features.id == 'auto_increment' then
		if not (#pk.parts == 1 and typeeq( pk.parts[1].type, 'num')) then
			error("For auto_increment numeric pk is mandatory",2+depth)
		end
		local pk_fno = pk.parts[1].fieldno
		gen_id = function()
			local max = pk:max()
			if max then
				return max[pk_fno] + 1
			else
				return 1
			end
		end
	elseif opts.features.id == 'time64' then
		if not (#pk.parts == 1 and typeeq( pk.parts[1].type, 'num')) then
			error("For time64 numeric pk is mandatory",2+depth)
		end
		gen_id = function()
			local key = clock.realtime64()
			while true do
				local exists = pk:get(key)
				if not exists then
					return key
				end
				key = key + 1
			end
		end
	elseif opts.features.id == 'uuid' then
		if not (#pk.parts == 1 and typeeq( pk.parts[1].type, 'str')) then
			error("For uuid string pk is mandatory",2+depth)
		end
		local uuid = require'uuid'
		gen_id = function()
			while true do
				local key = uuid.str()
				local exists = pk:get(key)
				if not exists then
					return key
				end
			end
		end
	elseif opts.features.id == 'required' then
		-- gen_id = function()
		-- 	-- ???
		-- 	error("TODO")
		-- end
	elseif type(opts.features.id) == 'function'
		or (debug.getmetatable(opts.features.id)
		and debug.getmetatable(opts.features.id).__call)
	then
		gen_id = opts.features.id
	else
		error(string.format(
			"Wrong type for features.id %s, may be 'auto_increment' | 'time64' | 'uuid' | 'required' | function",
			opts.features.id
		), 2+depth)
	end

	if not opts.features.retval then
		opts.features.retval = have_format and 'table' or 'tuple'
	end

	if format then
		self.tuple = _table2tuple(format)
		self.table = _tuple2table(format)
	end

	if opts.features.retval == 'table' then
		self.retwrap = self.table
	else
		self.retwrap = function(t) return t end
	end

	features.buried = not not opts.features.buried
	features.keep   = not not opts.features.keep
	if opts.features.delayed then
		if not have_runat then
			error(string.format("Delayed feature requires runat field and index" ),2+depth)
		end
		features.delayed = true
	else
		features.delayed = false
	end

	if opts.features.zombie then
		if features.keep then
			error(string.format("features keep and zombie are mutually exclusive" ),2+depth)
		end
		if not have_runat then
			error(string.format("feature zombie requires runat field and index" ),2+depth)
		end

		features.zombie = true
		if type(opts.features.zombie) == 'number' then
			features.zombie_delay = opts.features.zombie
		end
	else
		features.zombie = false
	end

	if opts.features.ttl then
		if not have_runat then
			error(string.format("feature ttl requires runat field and index" ),2+depth)
		end

		features.ttl = true
		if type(opts.features.ttl) == 'number' then
			features.ttl_default = opts.features.ttl
		end
	else
		features.ttl = false
	end

	if opts.features.ttr then
		if not have_runat then
			error(string.format("feature ttr requires runat field and index" ),2+depth)
		end

		features.ttr = true
		if type(opts.features.ttr) == 'number' then
			features.ttr_default = opts.features.ttr
		end
	else
		features.ttr = false
	end

	if fields.tube then
		features.tube = true
	end

	self.gen_id = gen_id
	self.features = features
	self.space = space.id

	function self.timeoffset(delta)
		delta = tonumber(delta) or 0
		return clock.realtime() + delta
	end
	function self.timeready(time)
		return time < clock.realtime()
	end
	function self.timeremaining(time)
		return time - clock.realtime()
	end
	-- self.NEVER = -1ULL
	self.NEVER = 0

	function self.keyfield(_,t)
		return t[pkf.no]
	end
	function self.keypack(_,t)
		return t[pkf.no]
	end

	-- Notify producer if it is still waiting us.
	-- Producer waits only for successfully processed task
	-- or for task which would never be processed.
	-- Discarding to process task depends on consumer's logic.
	-- Also task can be deleted from space by exceeding TTL.
	-- Producer should not be notified if queue is able to process task in future.
	local function notify_producer(key, task)
		local pkey = self:packkey(key)
		local wait = self.put_wait[pkey]
		if not wait then
			-- no producer
			return
		end

		if task.status == 'Z' or task.status == 'D' then
			-- task was processed
			wait.task = task
			wait.processed = true
			wait.cond:broadcast()
			return
		end
		if not space:get{key} then
			-- task is not present in space
			-- it could be TTL or :ack
			if task.status == 'T' then
				-- task was acked:
				wait.task = task
				wait.processed = true
				wait.cond:broadcast()
			elseif task.status == 'R' then
				-- task was killed by TTL
				wait.task = task
				wait.processed = false
				wait.cond:broadcast()
			end
		else
			-- task is still in space
			if task.status == 'B' then
				-- task was buried
				wait.task = task
				wait.processed = false
				wait.cond:broadcast()
			end
		end
	end

	self.ready = fiber.channel(0)

	local function rw_fiber_f(func, ...)
		local xq = self
		repeat
			if box.info.ro then
				log.verbose("awaiting rw")
				repeat
					if box.ctl.wait_rw then
						pcall(box.ctl.wait_rw, 1)
					else
						fiber.sleep(0.001)
					end
				until not box.info.ro
			end

			local ok, err = pcall(func, ...)
			if not ok then
				log.error("%s%s",
					err, xq.ready and ': (xq is not ready yet)' or '')
			end
			fiber.testcancel()
		until (not box.space[space.name]) or space.xq ~= xq
	end

	if opts.worker then
		local workers = opts.workers or 1
		local worker = opts.worker
		for i = 1,workers do
			fiber.create(rw_fiber_f, function(space,xq)
				local fname = space.name .. '.xq.wrk' .. tostring(i)
				---@diagnostic disable-next-line: undefined-field
				if package.reload then fname = fname .. '.' .. package.reload.count end
				fiber.name(string.sub(fname,1,32))
				repeat fiber.sleep(0.001) until space.xq
				if xq.ready then xq.ready:get() end
				log.info("I am worker %s",i)
				while box.space[space.name] and space.xq == xq and not box.info.ro do
					if xq.ready then xq.ready:get() end
					local task = space:take(1)
					if task then
						local key = xq:getkey(task)
						local r,e = pcall(worker,task)
						if not r then
							log.error("Worker for {%s} has error: %s", key, e)
						else
							if xq.taken[ key ] then
								space:ack(task)
							end
						end
						if xq.taken[ key ] then
							log.error("Worker for {%s} not released task", key)
							space:release(task)
						end
					end
					fiber.yield()
				end
				if box.info.ro then
					log.info("Shutting down on ro instance")
					return
				end
				log.info("worker %s ended", i)
			end,space,self)
		end
	end

	if have_runat then
		self.runat_chan = fiber.channel(0)
		self.runat = fiber.create(rw_fiber_f, function(space,xq,runat_index)
			local fname = space.name .. '.xq'
			if package.reload then fname = fname .. '.' .. package.reload.count end
			fiber.name(string.sub(fname,1,32))
			repeat fiber.sleep(0.001) until space.xq
			if xq.ready then xq.ready:get() end
			local chan = xq.runat_chan
			log.info("Runat started")
			local maxrun = 1000
			local curwait
			local collect = {}
			while box.space[space.name] and space.xq == xq and not box.info.ro do
				local r,e = pcall(function()
					-- print("runat loop 2 ",box.time64())
					local remaining
					for _,t in runat_index:pairs({0},{iterator = box.index.GT}) do

						-- print("checking ",t)
						if xq.timeready( t[ xq.fields.runat ] ) then
							table.insert(collect,t)
						else
							remaining = xq.timeremaining(t[ xq.fields.runat ])
							break
						end
						if #collect >= maxrun then remaining = 0 break end
					end

					for _,t in ipairs(collect) do
						-- log.info("Runat: %s, %s", _, t)
						if t[xq.fields.status] == 'W' then
							log.info("Runat: W->R %s",xq:keyfield(t))
							-- TODO: default ttl?
							local u = space:update({ xq:keyfield(t) },{
								{ '=',xq.fields.status,'R' },
								{ '=',xq.fields.runat, xq.NEVER }
							})
							xq:wakeup(u)
						elseif t[xq.fields.status] == 'R' and xq.features.ttl then
							local key = xq:keyfield(t)
							log.info("Runat: Kill R by ttl %s (%+0.2fs)", key, fiber.time() - t[ xq.fields.runat ])
							t = space:delete{key}
							notify_producer(key, t)
						elseif t[xq.fields.status] == 'Z' and xq.features.zombie then
							log.info("Runat: Kill Zombie %s",xq:keyfield(t))
							space:delete{ xq:keyfield(t) }
						elseif t[xq.fields.status] == 'T' and xq.features.ttr then
							local key = xq:keypack(t)
							local sid = xq.taken[ key ]
							local peer = peers[sid] or sid

							log.info("Runat: autorelease T->R by ttr %s taken by %s",xq:keyfield(t), peer)
							local u = space:update({ xq:keyfield(t) },{
								{ '=',xq.fields.status,'R' },
								{ '=',xq.fields.runat, xq.NEVER }
							})
							xq.taken[ key ] = nil
							if sid then
								self.bysid[ sid ][ key ] = nil
							end
							xq:wakeup(u)
						else
							log.error("Runat: unsupported status %s for %s",t[xq.fields.status], tostring(t))
							space:update({ xq:keyfield(t) },{
								{ '=',xq.fields.runat, xq.NEVER }
							})
						end
					end

					if remaining then
						if remaining >= 0 and remaining < 1 then
							return remaining
						end
					end
					return 1
				end)

				table_clear(collect)

				if r then
					curwait = e
				else
					curwait = 1
					log.error("Runat/ERR: %s",e)
				end
				-- log.info("Wait %0.2fs",curwait)
				if curwait == 0 then fiber.sleep(0) end
				chan:get(curwait)
			end
			if box.info.ro then
				log.info("Shutting down on ro instance")
				return
			end
			log.info("Runat ended")
		end,space,self,runat_index)
	end

	local function atomic_tail(self, key, status, ...)
		self._lock[key] = nil
		if not status then
			error((...), 3)
		end
		return ...
	end
	function self:atomic(key, fun, ...)
		self._lock[key] = true
		return atomic_tail(self, key, pcall(fun, ...))
	end

	function self:check_owner(key)
		local t = space:get(key)
		if not t then
			error(string.format( "Task {%s} was not found", key ),2)
		end
		if not self.taken[key] then
			error(string.format( "Task %s not taken by any", key ),2)
		end
		if self.taken[key] ~= box.session.id() then
			error(string.format( "Task %s taken by %d. Not you (%d)", key, self.taken[key], box.session.id() ),2)
		end
		return t
	end

	function self:putback(task)
		local key = task[ self.key.no ]
		local sid = self.taken[ key ]
		if sid then
			self.taken[ key ] = nil
			if self.bysid[ sid ] then
				self.bysid[ sid ][ key ] = nil
			else
				log.error( "Task {%s} marked as taken by sid=%s but bysid is null", key, sid)
			end
		else
			log.error( "Task {%s} not marked as taken, untake by sid=%s", key, box.session.id() )
		end

		notify_producer(key, task)
	end

	function self:wakeup(t)
		if t[self.fieldmap.status] ~= 'R' then return end
		if self.fieldmap.tube then
			-- we may have consumers in the tubes:
			local tube_chan = self.take_chans[t[self.fieldmap.tube]]
			if tube_chan and tube_chan:has_readers() and tube_chan:put(true, 0) then
				-- we have successfully notified consumer
				return
			end
			-- otherwise fallback to default channel:
		end
		if self.take_wait:has_readers() then
			self.take_wait:put(true,0)
		end
	end

	function self:make_ready()
		while self.ready and self.ready:has_readers() do
			self.ready:put(true,0)
			fiber.sleep(0)
		end
		self.ready = nil
	end

	local meta = debug.getmetatable(space)
	for k,v in pairs(methods) do meta[k] = v end

	-- Triggers must set right before updating space
	-- because raising error earlier leads to trigger inconsistency
	self._on_repl = space:on_replace(function(old, new)
		local old_st, new_st
		local old_tube, new_tube
		local old_tube_stat, new_tube_stat
		local counts = self._stat.counts
		local tube_ss = self._stat.tube

		if old then
			old_st = old[self.fields.status] --[[@as string]]
			if counts[old_st] and counts[old_st] > 0 then
				counts[old_st] = counts[old_st] - 1
			else
				log.error(
					"Have not valid statistic by status: %s with value: %s",
					old_st, tostring(counts[old_st])
				)
			end
			if self.fields.tube then
				old_tube = old[self.fields.tube]
				old_tube_stat = tube_ss[old_tube]
				if type(old_tube_stat) == 'table' and type(old_tube_stat.counts) == 'table' then
					if (tonumber(old_tube_stat.counts[old_st]) or 0) > 0 then
						old_tube_stat.counts[old_st] = old_tube_stat.counts[old_st] - 1
					else
						log.error(
							"Have not valid statistic by tubs/status: tube %s with value: %s",
							old_tube, old_st, tostring(tube_ss[old_tube].counts[old_st])
						)
					end
				else
					old_tube_stat = nil
				end
			end
		else
			old_st = 'X'
		end

		if new then
			new_st = new[self.fields.status] --[[@as string]]
			counts[new_st] = (counts[new_st] or 0LL) + 1
			if counts[new_st] < 0 then
				log.error("Statistic overflow by task type: %s", new_st)
			end
			if self.fields.tube then
				new_tube = new[self.fields.tube]
				new_tube_stat = tube_ss[new_tube]
				if type(new_tube_stat) == 'table' and type(new_tube_stat.counts) == 'table' then
					if (tonumber(new_tube_stat.counts[new_st]) or 0) >= 0 then
						new_tube_stat.counts[new_st] = (new_tube_stat.counts[new_st] or 0LL) + 1
					else
						log.error(
							"Have not valid statistic tube/status: tube %q with value: %s",
							new_tube, new_st, tostring(new_tube_stat.counts[new_st])
						)
					end
				else
					new_tube_stat = nil
				end
			end
		else
			new_st = 'X'
		end

		local field = old_st.."-"..new_st
		self._stat.transition[field] = (self._stat.transition[field] or 0LL) + 1
		if old_tube_stat then
			if not new_tube_stat or old_tube_stat == new_tube_stat then
				-- no new tube or new and old tubes are the same
				old_tube_stat.transition[field] = (old_tube_stat.transition[field] or 0LL) + 1
			else
				-- nil != old_tube != new_tube != nil
				-- cross tube transition ?
				-- Can this be backoff with tube change?
				local old_field = old_st.."-S"
				local new_field = "S-"..new_st
				old_tube_stat.transition[old_field] = (old_tube_stat.transition[old_field] or 0LL) + 1
				new_tube_stat.transition[new_field] = (new_tube_stat.transition[new_field] or 0LL) + 1
			end
		elseif new_tube_stat then
			-- old_tube_stat == nil
			new_tube_stat.transition[field] = (new_tube_stat.transition[field] or 0LL) + 1
		end
	end, self._on_repl)

	self._on_dis = box.session.on_disconnect(function()
		local sid = box.session.id()
		local peer = box.session.storage.peer

		log.info("%s: disconnected %s, sid=%s, fid=%s", space.name, peer, sid, fiber.id() )
		box.session.storage.destroyed = true
		if self.bysid[sid] then
			local old = self.bysid[sid]
			while next(old) do
				for key,realkey in pairs(old) do
					self.taken[key] = nil
					old[key] = nil
					local t = space:get(realkey)
					if t then
						if t[ self.fields.status ] == 'T' then
							self:wakeup(space:update({ realkey }, {
								{ '=',self.fields.status,'R' },
								self.have_runat and { '=', self.fields.runat, self.NEVER } or nil
							}))
							log.info("Rst: T->R {%s}", realkey )
						else
							log.error( "Rst: %s->? {%s}: wrong status", t[self.fields.status], realkey )
						end
					else
						log.error( "Rst: {%s}: taken not found", realkey )
					end
				end
			end
			self.bysid[sid] = nil
		end
	end, self._on_dis)

	rawset(space,'xq',self)

	log.info("Upgraded %s into xqueue (status=%s)", space.name, box.info.status)

	function self:starttest()
		local xq = self
		if box.info.status == 'orphan' then
			fiber.create(function()
				local x = 0
				repeat
					fiber.sleep(x/1e5)
					x = x + 1
				until box.info.status ~= 'orphan'
				self:starttest()
			end)
			return
		end
		if box.info.ro then
			log.info("Server is ro, not resetting statuses")
		else
			local space = box.space[self.space]
			local ver = tonumber(rawget(_G, '_TARANTOOL'):match('^([^.]+%.[^.]+)'))
			local yield_limit = 2^32-1
			local scanned = 0
			if ver >= 1.10 then
				yield_limit = 1000
			end
			box.begin()
			for _,t in self.index:pairs({'T'},{ iterator = box.index.EQ }) do
				scanned = scanned + 1
				local key = t[ xq.key.no ]
				if not self.taken[key] and not self._lock[key] then
					local update = {
						{ '=', self.fields.status, 'R' },
						self.have_runat and {
							'=',self.fields.runat,
							self.ttl_default and self.timeoffset( self.ttl_default ) or self.NEVER
						} or nil
					}
					space:update({key}, update)
					log.info("Start: T->R (%s)", key)
				end
				if scanned == yield_limit then
					scanned = 0
					box.commit()
					box.begin()
				end
			end
			box.commit()
		end
		self:make_ready()
	end
	self:starttest()
end


--[[
* `space:put`
	- `task` - table or array or tuple
		+ `table`
			* **requires** space format
			* suitable for id generation
		+ `array`
			* ignores space format
			* for id generation use `NULL` (**not** `nil`)
		+ `tuple`
			* ignores space format
			* **can't** be used with id generation
	- `attr` - table of attributes
		+ `delay` - number of seconds
			* if set, task will become `W` instead of `R` for `delay` seconds
		+ `ttl` - number of seconds
			* if set, task will be discarded after ttl seconds unless was taken
		+ `wait` - number of seconds
			* if set, callee fiber will be blocked up to `wait` seconds until task won't
			be processed or timeout reached.

```lua
box.space.myqueue:put{ name="xxx"; data="yyy"; }
box.space.myqueue:put{ "xxx","yyy" }
box.space.myqueue:put(box.tuple.new{ 1,"xxx","yyy" })

box.space.myqueue:put({ name="xxx"; data="yyy"; }, { delay = 1.5 })
box.space.myqueue:put({ name="xxx"; data="yyy"; }, { delay = 1.5; ttl = 100 })
```
]]

---@param t any[]|box.tuple
---@param opts? { delay: number?, ttl: number?, wait: number? }
---@return table|box.tuple, boolean? has_been_processed
function methods:put(t, opts)
	local xq = self.xq
	opts = opts or {}
	if type(t) == 'table' then
		if xq.gen_id then
			if is_array(t) then
				error("Task from array creation is not implemented yet", 2)
			else
				-- pass
			end
		end
	elseif type(t) == 'cdata' and ffi.typeof(t) == tuple_ctype then
		if xq.gen_id then
			error("Can't use id generation with tuple.", 2)
		end
		error("Task from tuple creation is not implemented yet", 2)
	else
		error("Wrong argument to put. Expected table or tuple", 2)
	end
	-- here we have table with keys
	if xq.gen_id then
		t[ xq.key.name ] = xq.gen_id()
	else
		if not t[ xq.key.name ] then
			error("Primary key is mandatory",2)
		end
	end

	-- delayed or ttl or default ttl
	if opts.delay then
		if not xq.features.delayed then
			error("Feature delayed is not enabled",2)
		end
		if opts.ttl then
			error("Features ttl and delay are mutually exclusive",2)
		end
		t[ xq.fieldmap.status ] = 'W'
		t[ xq.fieldmap.runat ]  = xq.timeoffset(opts.delay)

		if opts.wait then
			error("Are you crazy? Call of :put({...}, { wait = <>, delay = <> }) looks weird", 2)
		end
	elseif opts.ttl then
		if not xq.features.ttl then
			error("Feature ttl is not enabled",2)
		end
		t[ xq.fieldmap.status ] = 'R'
		t[ xq.fieldmap.runat ]  = xq.timeoffset(opts.ttl)
	elseif xq.features.ttl_default then
		t[ xq.fieldmap.status ] = 'R'
		t[ xq.fieldmap.runat ]  = xq.timeoffset(xq.features.ttl_default)
	elseif xq.have_runat then
		t[ xq.fieldmap.status ] = 'R'
		t[ xq.fieldmap.runat ]  = xq.NEVER
	else
		t[ xq.fieldmap.status ] = 'R'
	end

	local tuple = xq.tuple(t)
	local key = tuple[ xq.key.no ]

	local wait
	if opts.wait then
		wait = { cond = fiber.cond() }
		xq.put_wait[xq:packkey(key)] = wait
	end

	xq:atomic(key, function()
		t = self:insert(tuple)
	end)
	xq:wakeup(t)

	if wait then
		-- local func notify_producer will send to us some data
		local ok = wait.cond:wait(opts.wait)
		fiber.testcancel()
		if ok and wait.task then
			return xq.retwrap(wait.task), wait.processed
		end
	end
	return xq.retwrap(t)
end

--[[
* `space:wait(id, timeout)`
	- `id`:
		+ `string` | `number` - primary key
	- `timeout` - number of seconds to wait
		* callee fiber will be blocked up to `timeout` seconds until task won't
		be processed or timeout reached.
]]

local wait_for = {
	R = true,
	T = true,
	W = true,
}

---@param key table|scalar|box.tuple
---@param timeout number
---@return table|box.tuple, boolean? has_been_processed
function methods:wait(key, timeout)
	local xq = self.xq
	key = xq:getkey(key)
	local task = self:get(key)
	if not task then
		error(("Task {%s} was not found"):format(key))
	end

	local status = task[xq.fields.status]
	if not wait_for[status] then
		return xq.retwrap(task), false
	end

	local pkey = xq:packkey(key)
	local wait = xq.put_wait[pkey]
	if not wait then
		wait = { cond = fiber.cond() }
		xq.put_wait[pkey] = wait
	end

	-- local func notify_producer will send to us some data
	local ok = wait.cond:wait(timeout)
	fiber.testcancel()
	if ok and wait.task then
		return xq.retwrap(wait.task), wait.processed
	end

	return xq.retwrap(task), false
end

--[[
* `space:take(timeout)`
* `space:take(timeout, opts)`
* `space:take(opts)`
	- `timeout` - number of seconds to wait for new task
		+ choose reasonable time
		+ beware of **readahead** size (see tarantool docs)
	- `tube` - name of the tube worker wants to take task from (feature tube must be enabled)
	- returns task tuple or table (see retval) or nothing on timeout
	- *TODO*: ttr must be there
]]

---@param timeout? number|{ timeout: number?, ttr: number?, tube: string? }
---@param opts? { timeout: number?, ttr: number?, tube: string? }
---@return table|box.tuple?
function methods:take(timeout, opts)
	local xq = self.xq
	timeout = timeout or 0

	if type(timeout) == 'table' then
		opts = timeout
		timeout = opts.timeout or 0
	else
		opts = opts or {}
	end
	assert(timeout >= 0, "timeout required")

	local ttr
	if opts.ttr then
		if not xq.features.ttr then
			error("Feature ttr is not enabled",2)
		end
		ttr = opts.ttr
	elseif xq.features.ttr_default then
		ttr = xq.features.ttr_default
	end

	local index
	local start_with

	local tube_chan
	if opts.tube then
		if not xq.features.tube then
			error("Feature tube is not enabled", 2)
		end

		assert(type(opts.tube) == 'string', "opts.tube must be a string")

		index = xq.tube_index
		start_with = {opts.tube, 'R'}
		tube_chan = xq.take_chans[opts.tube] or fiber.channel()
		xq.take_chans[opts.tube] = tube_chan
	else
		index = xq.index
		start_with = {'R'}
	end
	---@cast index -nil

	local now = fiber.time()
	local key
	local found
	while not found do
		for _,t in index:pairs(start_with, { iterator = box.index.EQ }) do
			key = t[ xq.key.no ]
			if not xq._lock[ key ] then
				-- found key
				xq._lock[ key ] = true
				found = t
				break
			end
		end
		if not found then
			local left = (now + timeout) - fiber.time()
			if left <= 0 then goto finish end

			(tube_chan or xq.take_wait):get(left)
			if box.session.storage.destroyed then goto finish end
		end
	end
	::finish::

	-- If we were last reader from the tube
	-- we remove channel. Writers will get false
	-- on :put if they exists.
	if tube_chan and not tube_chan:has_readers() then
		tube_chan:close()
		xq.take_chans[opts.tube] = nil
	end
	if not found then return end

	local r,e = pcall(function()
		local sid = box.session.id()
		local peer = box.session.storage.peer

		-- print("Take ",key," for ",peer," sid=",sid, "; fid=",fiber.id() )
		if xq.debug then
			log.info("Take {%s} by %s, sid=%s, fid=%s", key, peer, sid, fiber.id())
		end

		local update = {
			{ '=', xq.fields.status, 'T' },
		}
		-- TODO: update more fields

		if ttr then
			table.insert(update,{ '=', xq.fields.runat, xq.timeoffset(ttr)})
			xq.runat_chan:put(true,0)
		end
		local t = self:update({key},update)

		if not xq.bysid[ sid ] then xq.bysid[ sid ] = {} end
		xq.taken[ key ] = sid
		xq.bysid[ sid ][ key ] = key

		return t
	end)

	xq._lock[ key ] = nil
	if not r then
		error(e)
	end
	return xq.retwrap(e)
end

--[[
* `space:release(id, [attr])`
	- `id`:
		+ `string` | `number` - primary key
		+ *TODO*: `tuple` - key will be extracted using index
		+ *TODO*: composite pk
	- `attr`
		+ `update` - table for update, like in space:update
		+ `ttl` - timeout for time to live
		+ `delay` - number of seconds
			* if set, task will become `W` instead of `R` for `delay` seconds
]]

---@param key table|scalar|box.tuple
---@param attr? {delay: number?, ttl: number?, update: { [1]: update_operation, [2]: number|string, [3]: tuple_type }[] }
---@return table|box.tuple
function methods:release(key, attr)
	local xq = self.xq
	key = xq:getkey(key)

	attr = attr or {}

	local t = xq:check_owner(key)
	local old = t[ xq.fields.status ]
	local update = {}
	if attr.update then
		for _,v in pairs(attr.update) do table.insert(update,v) end
	end

	-- delayed or ttl or default ttl
	if attr.delay then
		if not xq.features.delayed then
			error("Feature delayed is not enabled",2)
		end
		if attr.ttl then
			error("Features ttl and delay are mutually exclusive",2)
		end
		table.insert(update, { '=', xq.fields.status, 'W' })
		table.insert(update, { '=', xq.fields.runat, xq.timeoffset(attr.delay) })
	elseif attr.ttl then
		if not xq.features.ttl then
			error("Feature ttl is not enabled",2)
		end
		table.insert(update, { '=', xq.fields.status, 'R' })
		table.insert(update, { '=', xq.fields.runat, xq.timeoffset(attr.ttl) })
	elseif xq.features.ttl_default then
		table.insert(update, { '=', xq.fields.status, 'R' })
		table.insert(update, { '=', xq.fields.runat, xq.timeoffset(xq.features.ttl_default) })
	elseif xq.have_runat then
		table.insert(update, { '=', xq.fields.status, 'R' })
		table.insert(update, { '=', xq.fields.runat, xq.NEVER })
	else
		table.insert(update, { '=', xq.fields.status, 'R' })
	end

	xq:atomic(key,function()
		t = self:update({key}, update)

		---@cast t box.tuple
		xq:wakeup(t)
		if xq.have_runat then
			xq.runat_chan:put(true,0)
		end

		log.info("Rel: %s->%s {%s} +%s from %s/sid=%s/fid=%s", old, t[xq.fields.status],
			key, attr.delay, box.session.storage.peer, box.session.id(), fiber.id() )
	end)

	xq:putback(t)

	return t
end

---@param key table|scalar|box.tuple
---@param attr? {delay: number?, ttl: number?, update: { [1]: update_operation, [2]: number|string, [3]: tuple_type }[] }
---@return table|box.tuple
function methods:ack(key, attr)
	-- features.zombie
	-- features.keep
	local xq = self.xq
	key = xq:getkey(key)

	attr = attr or {}

	local t = xq:check_owner(key)
	local old = t[ xq.fields.status ]
	local delete = false
	local update = {}
	if attr.update then
		for _,v in pairs(attr.update) do table.insert(update,v) end
	end

	-- delayed or ttl or default ttl
	if attr.delay then
		if not xq.features.zombie then
			error("Feature zombie is not enabled",2)
		end
		table.insert(update, { '=', xq.fields.status, 'Z' })
		table.insert(update, { '=', xq.fields.runat, xq.timeoffset(attr.delay) })
	elseif xq.features.zombie then
		table.insert(update, { '=', xq.fields.status, 'Z' })
		if xq.features.zombie_delay then
			table.insert(update, { '=', xq.fields.runat, xq.timeoffset(xq.features.zombie_delay) })
		end
	elseif xq.features.keep then
		table.insert(update, { '=', xq.fields.status, 'D' })
	else
		-- do remove task
		delete = true
	end

	xq:atomic(key,function()
		if #update > 0 then
			t = self:update({key}, update)
			---@cast t box.tuple
			xq:wakeup(t)
			if xq.have_runat then
				xq.runat_chan:put(true,0)
			end
			log.info("Ack: %s->%s {%s} +%s from %s/sid=%s/fid=%s", old, t[xq.fields.status],
				key, attr.delay, box.session.storage.peer, box.session.id(), fiber.id() )
		end

		if delete then
			t = self:delete{key}
			log.info("Ack: %s->delete {%s} +%s from %s/sid=%s/fid=%s", old,
				key, attr.delay, box.session.storage.peer, box.session.id(), fiber.id() )
		end
	end)

	xq:putback(t) -- in real drop form taken key

	return t
end

---@param key table|scalar|box.tuple
---@param attr? {delay: number?, ttl: number?, update: { [1]: update_operation, [2]: number|string, [3]: tuple_type }[] }
function methods:bury(key, attr)
	attr = attr or {}

	local xq = self.xq
	key = xq:getkey(key)
	local t = xq:check_owner(key)

	local update = {}
	if attr.update then
		for _,v in pairs(attr.update) do table.insert(update,v) end
	end
	table.insert(update, { '=', xq.fields.status, 'B' })

	xq:atomic(key,function()
		t = self:update({key}, update)

		xq:wakeup(t)
		if xq.have_runat then
			xq.runat_chan:put(true,0)
		end
		log.info("Bury {%s} by %s, sid=%s, fid=%s", key, box.session.storage.peer, box.session.id(), fiber.id())
	end)

	xq:putback(t)
end

local function kick_task(self, key, attr)
	local xq   = self.xq
	key  = xq:getkey(key)
	local peer = box.session.storage.peer
	local sid  = box.session.id()
	attr       = attr or {}

	if xq.debug then
		log.info("Kick {%s} by %s, sid=%s, fid=%s", key, peer, sid, fiber.id())
	end

	self:update({key}, {{ '=', xq.fields.status, 'R' }})
	xq:putback(key, attr)
end

function methods:kick(nr_tasks_or_task, attr)
	attr = attr or {}
	if type(nr_tasks_or_task) == 'number' then
		local task_n = 1
		for _, t in self.xq.index:pairs({'B'},{ iterator = box.index.EQ }) do
			if task_n > nr_tasks_or_task then break end
			kick_task(self, t, attr)
			if task_n % 500 == 0 then fiber.sleep(0) end
			task_n = task_n + 1
		end
	else
		kick_task(self, nr_tasks_or_task, attr)
	end
end

---@param key table|scalar|box.tuple
---@return box.tuple|table
function methods:kill(key)
	local xq     = self.xq
	key = xq:getkey(key)
	local task   = self:get(key)
	if not task then
		error(("Task by %s not found to kill"):format(key))
	end
	local peer   = box.session.storage.peer

	if xq.debug then
		log.info("Kill {%s} by %s, sid=%s, fid=%s", key, peer, box.session.id(), fiber.id())
	end

	task = self:delete(key)
	---@cast task -nil
	---Kill is treated as a synonim of Bury
	task = task:update({{ '=', xq.fields.status, 'B' }})
	xq:putback(task)
	return xq.retwrap(task)
end

-- special remap of truncate for deliting stats and saving methods
function methods:truncate()
	local xq = self.xq
	xq.ready = xq.ready or fiber.channel(0)
	local stat = xq._stat
	for status, _ in pairs(stat.counts) do
		stat.counts[status] = 0LL
	end
	for transition, _ in pairs(stat.transition) do
		stat.transition[transition] = nil
	end
	local ret = xq._default_truncate(self)
	local meta = debug.getmetatable(self)
	for k,v in pairs(methods) do meta[k] = v end
	xq:make_ready()
	-- Now we reset our methods after truncation because
	-- as we can see in on_replace_dd_truncate:
	-- https://github.com/tarantool/tarantool/blob/0b7cc52607b2290d2f35cc68ee1a8243988c2735/src/box/alter.cc#L2239
	-- tarantool deletes space and restores it with the same indexes
	-- but without methods
	return ret
end

local shortmap = { __serialize = 'map' }

---@param pretty? boolean
function methods:stats(pretty)
	local stats = table.deepcopy(self.xq._stat)
		for s, ps in pairs(pretty_st) do
			if pretty then
				stats.counts[ps] = stats.counts[s] or 0LL
				stats.counts[s]  = nil
				for _, tube_stat in pairs(stats.tube) do
					tube_stat.counts[ps] = tube_stat.counts[s] or 0LL
					tube_stat.counts[s] = nil
				end
			else
				stats.counts[s] = stats.counts[s] or 0LL
				for _, tube_stat in pairs(stats.tube) do
					setmetatable(tube_stat.counts, shortmap)
					tube_stat.counts[s] = tube_stat.counts[s] or 0LL
				end
			end
		end
		setmetatable(stats.counts, shortmap)
	return stats
end

setmetatable(M,{
	__call = function(_, space, opts)
		M.upgrade(space,opts,1)
	end
})

return M