adds annotations to xqueue

Author: ochaton

.luacheckrc

@@ -1,4 +1,5 @@
 std = "luajit"
+codes = true
 globals = {
 	-- Tarantool variable:
 	"box",
@@ -10,8 +11,11 @@ globals = {
 }
 
 ignore = {
--- 	"211",
+	"211",
+	"212",
+	"431",
+	"432",
+	"542",
 	"611",
 -- 	"631",
--- 	"542"
 }

.vscode/settings.json

@@ -0,0 +1,3 @@
+{
+	"Lua.runtime.version": "LuaJIT"
+}

xqueue.lua

@@ -74,7 +74,7 @@ Status:
 		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
@@ -84,7 +84,7 @@ Status:
 
 	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)
@@ -219,8 +219,70 @@ local function _table2tuple ( qformat )
 	return dostring(fun)
 end
 
+---@type xqueueSpace
 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()
+
+
+---@class xqueueSpace: boxSpaceObject
+---@field xq xq xqueue specific storage
+
+---Upgrades given space to xqueue instance
+---@param space xqueueSpace
+---@param opts table
+---@param depth? number
 function M.upgrade(space,opts,depth)
 	depth = depth or 0
 	log.info("xqueue upgrade(%s,%s)", space.name, json.encode(opts))
@@ -269,11 +331,11 @@ function M.upgrade(space,opts,depth)
 		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
@@ -365,7 +427,7 @@ function M.upgrade(space,opts,depth)
 	self.key = pkf
 	self.fields = fields
 	self.fieldmap = fieldmap
-	
+
 	if not self._stat then
 		self._stat = {
 			counts = {};
@@ -376,7 +438,7 @@ function M.upgrade(space,opts,depth)
 			self._stat.counts[s] = (self._stat.counts[s] or 0LL) + 1
 		end
 	end
-	
+
 	function self:getkey(arg)
 		local _type = type(arg)
 		if _type == 'table' then
@@ -459,6 +521,7 @@ function M.upgrade(space,opts,depth)
 		end
 	end
 
+	---@type boxIndex
 	local runat_index
 	if fields.runat then
 		for _,index in pairs(space.index) do
@@ -768,7 +831,7 @@ function M.upgrade(space,opts,depth)
 						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
@@ -817,7 +880,7 @@ function M.upgrade(space,opts,depth)
 					end
 					return 1
 				end)
-				
+
 				table_clear(collect)
 
 				if r then
@@ -900,10 +963,10 @@ function M.upgrade(space,opts,depth)
 		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)
@@ -922,7 +985,7 @@ function M.upgrade(space,opts,depth)
 		else
 			old_st = 'X'
 		end
-		
+
 		if new then
 			new_st = new[self.fields.status]
 			counts[new_st] = (counts[new_st] or 0LL) + 1
@@ -932,15 +995,15 @@ function M.upgrade(space,opts,depth)
 		else
 			new_st = 'X'
 		end
-		
+
 		local field = old_st.."-"..new_st
 		self._stat.transition[field] = (self._stat.transition[field] or 0ULL) + 1
 	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
@@ -968,7 +1031,7 @@ function M.upgrade(space,opts,depth)
 			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)
@@ -1263,7 +1326,7 @@ function methods:take(timeout, opts)
 	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())
@@ -1350,6 +1413,7 @@ function methods:release(key, attr)
 	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)
@@ -1358,9 +1422,9 @@ function methods:release(key, attr)
 		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
 
@@ -1402,6 +1466,7 @@ function methods:ack(key, attr)
 	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)
@@ -1450,7 +1515,7 @@ end
 
 local function kick_task(self, key, attr)
 	local xq   = self.xq
-	local key  = xq:getkey(key)
+	key  = xq:getkey(key)
 	local peer = box.session.storage.peer
 	local sid  = box.session.id()
 	attr       = attr or {}
@@ -1480,7 +1545,7 @@ end
 
 function methods:kill(key)
 	local xq     = self.xq
-	local key    = xq:getkey(key)
+	key    = xq:getkey(key)
 	local task   = self:get(key)
 	local status = task[xq.fields.status]
 	local peer   = box.session.storage.peer