drivers: fix utube kick in the ready buffer mode

CHANGELOG.md

@@ -13,6 +13,14 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ### Fixed
 
+- `kick()` corrupted the ready buffer of the `utube` driver in the
+  `ready_buffer` storage mode. A kicked task either stayed invisible to
+  `take()` forever, or was written to the buffer under its status instead of
+  its utube name, which allowed two tasks of one utube to be taken at once
+  (#256).
+- `kick()` left a transaction open when the `utube` tube had no buried tasks
+  (#256).
+
 ## [1.5.0] - 2026-07-10
 
 This release introduces tube-level grants for roles (supported by all default

queue/abstract/driver/utube.lua

@@ -179,6 +179,19 @@ local function put_ready(self, id, utube)
     end
 end
 
+-- Try to update the current task in ready_buffer for the given 'utube'.
+local function update_ready(self, id, utube)
+    local prev_task = self.space_ready_buffer.index.utube:get{utube}
+    if prev_task ~= nil then
+        if prev_task[1] > id then
+            self.space_ready_buffer:delete(prev_task[1])
+            self.space_ready_buffer:insert({id, utube})
+        end
+    else
+        put_ready(self, id, utube)
+    end
+end
+
 local function commit()
     box.commit()
 end
@@ -413,23 +426,19 @@ function method.kick(self, count)
 
         local task = self.space.index.status:min{ state.BURIED }
         if task == nil then
+            commit_func()
+
             return i - 1
         end
         if task[2] ~= state.BURIED then
+            commit_func()
+
             return i - 1
         end
 
         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
+            update_ready(self, task[1], task[3])
         end
 
         commit_func()

t/260-utube-kick.t

@@ -0,0 +1,86 @@
+#!/usr/bin/env tarantool
+local test = (require('tap')).test()
+test:plan(2)
+
+local tnt = require('t.tnt')
+tnt.cfg{}
+
+local engine = os.getenv('ENGINE') or 'memtx'
+
+local queue = require('queue')
+
+-- The 'ready_buffer' storage mode is not supported by the vinyl engine.
+test:test('kick with the ready buffer', function(test)
+    if engine == 'vinyl' then
+        test:plan(1)
+        test:ok(true, 'skipped for vinyl')
+        return
+    end
+    test:plan(4)
+
+    local buffer_name = 'kick_ready_buffer'
+
+    local tube = queue.create_tube(buffer_name, 'utube', {
+        engine = engine,
+        storage_mode = queue.driver.utube.STORAGE_MODE_READY_BUFFER
+    })
+    local buffer = box.space[buffer_name .. '_ready_buffer']
+
+    -- A kicked task must be returned to the ready buffer, otherwise it stays
+    -- READY but is never seen by take().
+    local task = tube:put('single', {utube = 'u'})
+    tube:bury(task[1])
+    test:is(tube:kick(1), 1, 'the buried task is kicked')
+
+    local function select_buffer(...)
+        local rows = {}
+        for _, tuple in ipairs(buffer:select(...)) do
+            table.insert(rows, tuple:totable())
+        end
+        return rows
+    end
+
+    test:is_deeply(select_buffer(), {{task[1], 'u'}},
+        'the kicked task is in the ready buffer under its utube name')
+
+    local taken = tube:take(0)
+    test:is(taken and taken[1], task[1], 'the kicked task can be taken')
+    if taken ~= nil then
+        tube:ack(taken[1])
+    else
+        tube:truncate()
+    end
+
+    -- A kicked task that displaces a younger task of the same utube must be
+    -- written to the buffer with the utube name, not with its status.
+    local first  = tube:put('first',  {utube = 'v'})
+    local second = tube:put('second', {utube = 'v'})
+    tube:take(0)             -- takes 'first', drops it from the buffer
+    tube:bury(first[1])      -- buries it, 'second' takes its buffer slot
+    tube:kick(1)             -- kicks 'first' back, it displaces 'second'
+
+    local rows = {}
+    for _, tuple in ipairs(buffer.index.utube:select({'v'})) do
+        table.insert(rows, tuple:totable())
+    end
+    test:is_deeply(rows, {{first[1], 'v'}},
+        'the displacing task keeps the utube name in the buffer')
+end)
+
+test:test('kick does not leak a transaction', function(test)
+    test:plan(1)
+
+    local tube = queue.create_tube('kick_txn', 'utube', {engine = engine})
+
+    tube:kick(1)  -- nothing is buried, kick returns early
+
+    local leaked = box.is_in_txn()
+    if leaked then
+        box.rollback()
+    end
+    test:ok(not leaked, 'kick over an empty tube leaves no open transaction')
+end)
+
+tnt.finish()
+os.exit(test:check() and 0 or 1)
+-- vim: set ft=lua :