drivers: unable to take a task with utubettl

Previously, the function incorrectly retrieved the lowest priority
task instead of the actual task in the ready buffer. This caused a
logical error when the current task had the lowest priority.

The implementation error caused:

1. Incorrect selection of the lowest priority task for a take.
2. ready buffer appearing empty when tasks were actually in ready
   state.

Author: oleg-jukovec

CHANGELOG.md

@@ -13,6 +13,10 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ### Fixed
 
+- Incorrect choice by priority of task with utubettl driver + ready buffer
+  mode (#244).
+- Unable to take task with utubettl driver + ready buffer mode (#244).
+
 ## [1.4.3] - 2024-03-05
 
 The release fixes start of a queue on instances with gaps inside

queue/abstract/driver/utubettl.lua

@@ -137,14 +137,17 @@ end
 local function put_ready(self, id, utube, pri)
     local taken = self.space.index.utube:min{state.TAKEN, utube}
     if taken == nil or taken[i_status] ~= state.TAKEN then
-        local current_task = self.space.index.utube_pri:min{state.READY, utube}
-        if current_task[i_status] ~= state.READY or
-                current_task[i_pri] < pri or (current_task[i_pri] == pri and current_task[i_id] < id) then
-            return
-        end
-        if current_task[i_pri] > pri then
-            self.space_ready_buffer:delete(current_task[id])
+        local cur_task = self.space_ready_buffer.index.utube:get{utube}
+
+        if cur_task ~= nil then
+            local cur_id = cur_task[1]
+            local cur_pri = cur_task[3]
+            if cur_pri < pri or (cur_pri == pri and cur_id < id) then
+                return
+            end
+            self.space_ready_buffer:delete(cur_id)
         end
+
         -- Ignoring ER_TUPLE_FOUND error, if a tuple with the same task_id
         -- or utube name is already in the space.
         -- Note that both task_id and utube indexes are unique, so there will be

t/040-utubettl.t

@@ -3,7 +3,7 @@ local yaml  = require('yaml')
 local fiber = require('fiber')
 
 local test = (require('tap')).test()
-test:plan(17)
+test:plan(18)
 
 local queue = require('queue')
 local state = require('queue.abstract.state')
@@ -208,6 +208,7 @@ test:test('bury in utube', function(test)
         test:is(state, 2, 'state was changed')
     end
 end)
+
 test:test('instant bury', function(test)
     if engine ~= 'vinyl' then
         test:plan(1 * 2)
@@ -225,6 +226,40 @@ test:test('instant bury', function(test)
         test:is(tube_ready:bury(taken[1])[2], '!', 'task is buried')
     end
 end)
+
+test:test('priority in utube', function(test)
+    if engine ~= 'vinyl' then
+        test:plan(8 * 2)
+    else
+        test:plan(8)
+    end
+
+    for _, test_tube in ipairs({tube, tube_ready}) do
+        if test_tube == nil then
+            break
+        end
+
+        test:ok(test_tube:put(670, {utube = 'dee', pri = 1}), 'task was put')
+        test:ok(test_tube:put(671, {utube = 'dee', pri = 0}), 'task was put')
+
+        local taken = test_tube:take(.1)
+        test:ok(taken, 'task was taken ' .. taken[1])
+        test:is(taken[3], 671, 'task.data')
+
+        test_tube:release(taken[1])
+
+        taken = test_tube:take(.1)
+        test:ok(taken, 'task was taken ' .. taken[1])
+        test:is(taken[3], 671, 'task.data')
+        test_tube:ack(taken[1])
+
+        taken = test_tube:take(.1)
+        test:ok(taken, 'task was taken ' .. taken[1])
+        test:is(taken[3], 670, 'task.data')
+        test_tube:ack(taken[1])
+    end
+end)
+
 test:test('release in utube', function(test)
     if engine ~= 'vinyl' then
         test:plan(8 * 2)