Infinite loop on (possible) ttr return of a buried task to the dropped queue

[DifferentialOrange]

Tarantool Enterprise 1.10.3-89-g412e943, queue 1.0.3

Bug can be reproduced by the following script

local queue = require('queue')
local fiber = require('fiber')

box.cfg{}
queue.create_tube('task_line', 'fifottl', {ttr = 10, if_not_exist = true})
queue.tube.task_line:put({task = "task"})
queue.tube.task_line:put({task1 = "task"})
local taken_task = queue.tube.task_line:take(0)

local tube_id = box.space._queue:select{tube_name = 'task_line'}[1].tube_id
local tasks_taken = box.space._queue_taken:select{tube_id = tube_id}

for _, task in ipairs(tasks_taken) do
    queue.tube.task_line:bury(task.task_id)
end
queue.tube.task_line:drop()
fiber.sleep(20)

After ~ttr seconds of sleep tarantool instance start to use ~100% of CPU, became unavailable for connection by net.box and can't be stopped by Ctrl-C.

[DifferentialOrange]

/queue/abstract/driver/fifottl.lua

fifottl_fiber_iteration function throws error on
task = self.space.index.watch:min(delayed_state) with following
main/112/fifottl I> ....rocks/share/tarantool/queue/abstract/driver/fifottl.lua:102: attempt to index field 'watch' (a nil value)

On fifottl_fiber cycle

while true do
        if not box.cfg.read_only then
                local stat, err = pcall(fifottl_fiber_iteration, self, processed)
                if not stat and not err.code == box.error.READONLY then
                        log.error("error catched: %s", tostring(err))
                        log.error("exiting fiber '%s'", fiber.name())
                return 1
                elseif stat then
                        processed = err
                end
        else
                fiber.sleep(0.1)
        end
end

after pcall script do not satisfy if and elseif condition, goes again on pcall and do not yield.

[opomuc]

https://github.com/tarantool/queue/blob/master/queue/abstract/driver/fifottl.lua#L101