utubettl-on-task-change: commit after on_task_change

Author: Алексей Зорькин

CHANGELOG.md

@@ -13,6 +13,15 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ### Fixed
 
+## [1.4.5] - 2025-09-22
+
+Race condition between utubettl_fiber_iteration fiber and other operation.
+
+### Fixed
+
+- Erasing the task entry from _queue_taken_2 in the `utubettl` driver by
+  a race condition between utubettl_fiber_iteration fiber and other operation.
+
 ## [1.4.4] - 2025-05-26
 
 The patch release fixes incorrect behavior of the utubettl driver with enabled

queue/abstract.lua

@@ -62,7 +62,7 @@ local function tube_release_all_orphaned_tasks(tube)
         }
         if taken and session.exist_shared(taken[4]) then
             log.info(prefix ..
-                ('skipping task: %s, tube_id: %d'):format(task[1],
+                ('skipping task: %d, tube_id: %d'):format(task[1],
                     tube.tube_id))
         else
             tube.raw:release(task[1], {})

queue/abstract/driver/utubettl.lua

@@ -207,7 +207,6 @@ local function utubettl_fiber_iteration(self, processed)
     local estimated = util.MAX_TIMEOUT
 
     local commit_func = begin_if_not_in_txn(self)
-    local commited = false
 
     -- delayed tasks
     task = self.space.index.watch:min(delayed_state)
@@ -221,22 +220,21 @@ local function utubettl_fiber_iteration(self, processed)
             if self.ready_space_mode then
                 update_ready(self, task[i_id], task[i_utube], task[i_pri])
             end
-            self:on_task_change(task, 'delayed')
-            commit_func()
-            commited = true
 
+            self:on_task_change(task, 'delayed')
             estimated = 0
             processed = processed + 1
         else
             estimated = tonumber(task[i_next_event] - now) / 1000000
         end
     end
-    if not commited then
-        commit_func()
-    end
+
+    commit_func()
 
     -- ttl tasks
     for _, state in pairs(ttl_states) do
+        commit_func = begin_if_not_in_txn(self)
+
         task = self.space.index.watch:min{ state }
         if task ~= nil and task[i_status] == state then
             if now >= task[i_next_event] then
@@ -249,10 +247,12 @@ local function utubettl_fiber_iteration(self, processed)
                 estimated = et < estimated and et or estimated
             end
         end
+
+        commit_func()
     end
 
     commit_func = begin_if_not_in_txn(self)
-    commited = false
+
     -- ttr tasks
     task = self.space.index.watch:min(ttr_state)
     if task and task[i_status] == state.TAKEN then
@@ -265,20 +265,17 @@ local function utubettl_fiber_iteration(self, processed)
             if self.ready_space_mode then
                 put_ready(self, task[i_id], task[i_utube], task[i_pri])
             end
-            self:on_task_change(task, 'ttr')
-            commit_func()
-            commited = true
 
+            self:on_task_change(task, 'ttr')
             estimated = 0
             processed = processed + 1
         else
             local et = tonumber(task[i_next_event] - now) / 1000000
             estimated = et < estimated and et or estimated
         end
     end
-    if not commited then
-        commit_func()
-    end
+
+    commit_func()
 
     if estimated > 0 or processed > 1000 then
         -- free refcounter
@@ -450,9 +447,9 @@ function method.put(self, data, opts)
         put_ready(self, task[i_id], task[i_utube], task[i_pri])
     end
 
-    self:on_task_change(task, 'put')
     commit_func()
 
+    self:on_task_change(task, 'put')
     return task
 end
 
@@ -515,13 +512,13 @@ local function take_ready(self)
             end
         end
 
+        commit_func()
+
         if take_complete then
             self:on_task_change(task, 'take')
-            commit_func()
             return task
         elseif take_ttl then
             self:on_task_change(task, 'ttl')
-            commit_func()
         end
     end
 end
@@ -554,13 +551,11 @@ local function take(self)
                 take_complete = true
             end
 
+            commit_func()
             if take_complete then
                 self:on_task_change(t, 'take')
-                commit_func()
                 return t
             end
-
-            commit_func()
         end
     end
 end
@@ -609,13 +604,11 @@ function method.delete(self, id)
         commit_func()
 
         if is_taken then
-            task = process_neighbour(self, task, 'delete')
+            return process_neighbour(self, task, 'delete')
         else
             self:on_task_change(task, 'delete')
+            return task
         end
-
-        commit_func()
-        return task
     end
 
     commit_func()
@@ -641,10 +634,9 @@ function method.release(self, id, opts)
                 put_next_ready(self, task[i_utube])
             end
 
-            task = process_neighbour(self, task, 'release')
             commit_func()
 
-            return task
+            return process_neighbour(self, task, 'release')
         end
     else
         task = self.space:update(id, {
@@ -657,8 +649,8 @@ function method.release(self, id, opts)
         end
     end
 
-    self:on_task_change(task, 'release')
     commit_func()
+    self:on_task_change(task, 'release')
     return task
 end
 
@@ -689,12 +681,11 @@ function method.bury(self, id)
         end
     end
 
-    task = process_neighbour(
-            self, task:transform(i_status, 1, state.BURIED), 'bury'
-    )
     commit_func()
 
-    return task
+    return process_neighbour(
+        self, task:transform(i_status, 1, state.BURIED), 'bury'
+    )
 end
 
 -- unbury several tasks
@@ -719,8 +710,9 @@ function method.kick(self, count)
             update_ready(self, task[i_id], task[i_utube], task[i_pri])
         end
 
-        self:on_task_change(task, 'kick')
         commit_func()
+
+        self:on_task_change(task, 'kick')
     end
     return count
 end