Two defects in method.kick() of the utube driver, both only in the STORAGE_MODE_READY_BUFFER mode. utubettl does the same thing correctly via its update_ready() helper.
|
task = self.space:update(task[1], {{ '=', 2, state.READY }}) |
|
if self.ready_space_mode then |
|
local prev_task = self.space_ready_buffer.index.utube:get{task[3]} |
|
if prev_task ~= nil then |
|
if prev_task[1] > task[1] then |
|
self.space_ready_buffer:delete(prev_task[1]) |
|
self.space_ready_buffer:insert({task[1], task[2]}) |
|
end |
|
else |
|
put_ready(self, task[3]) |
|
end |
|
end |
1. Wrong argument list. When the utube has no entry in the ready buffer, kick() calls put_ready(self, task[3]), while the helper is put_ready(self, id, utube). The utube name arrives as the task id and the utube becomes nil, so the insert fails on the space format and the error is swallowed by the pcall inside put_ready(). The kicked task stays READY but is never returned by take(), because take_ready() only looks at the ready buffer.
local tube = queue.create_tube('t', 'utube',
{storage_mode = queue.driver.utube.STORAGE_MODE_READY_BUFFER})
local task = tube:put('single', {utube = 'u'})
tube:bury(task[1])
tube:kick(1) -- returns 1
box.space.t_ready_buffer:select() -- [] , expected [[0, 'u']]
tube:take(0) -- nil, the task is stuck
2. Status written instead of the utube name. When the utube already has a younger task in the buffer, kick() replaces it with insert({task[1], task[2]}), that is {id, status} instead of {id, utube}. The status string lands in the utube field, so the buffer no longer has an entry under the real utube name and a later put() adds a second one. With two entries for one utube the driver can take two tasks of the same utube at once, which breaks the main utube guarantee, and take_ready() can spin forever on an entry whose utube already has a TAKEN task.
3. Leaked transaction. Both early returns of kick() return without calling commit_func(), so with no buried tasks the transaction opened by begin_if_not_in_txn() stays running and leaks into whatever the caller fiber does next:
tube:kick(1) -- nothing is buried
box.is_in_txn() -- true
utubettl commits before both early returns. This part is within the scope of #231, the rest is not.
Two defects in
method.kick()of theutubedriver, both only in theSTORAGE_MODE_READY_BUFFERmode.utubettldoes the same thing correctly via itsupdate_ready()helper.queue/queue/abstract/driver/utube.lua
Lines 422 to 433 in 03cece7
1. Wrong argument list. When the utube has no entry in the ready buffer,
kick()callsput_ready(self, task[3]), while the helper isput_ready(self, id, utube). The utube name arrives as the task id and the utube becomesnil, so the insert fails on the space format and the error is swallowed by thepcallinsideput_ready(). The kicked task stays READY but is never returned bytake(), becausetake_ready()only looks at the ready buffer.2. Status written instead of the utube name. When the utube already has a younger task in the buffer,
kick()replaces it withinsert({task[1], task[2]}), that is{id, status}instead of{id, utube}. The status string lands in the utube field, so the buffer no longer has an entry under the real utube name and a laterput()adds a second one. With two entries for one utube the driver can take two tasks of the same utube at once, which breaks the main utube guarantee, andtake_ready()can spin forever on an entry whose utube already has a TAKEN task.3. Leaked transaction. Both early returns of
kick()return without callingcommit_func(), so with no buried tasks the transaction opened bybegin_if_not_in_txn()stays running and leaks into whatever the caller fiber does next:utubettlcommits before both early returns. This part is within the scope of #231, the rest is not.