utubettl: remove extra sleep from utubettl_fiber

CHANGELOG.md

@@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Fixed
+
+- Too long timings in the `utubettl` driver (#223)
+
 ## [1.4.0] - 2024-05-20
 
 The release introduces an experimental `storage_mode` option for the `utube`

queue/abstract/driver/utubettl.lua

@@ -305,11 +305,12 @@ local function utubettl_fiber(self)
             elseif stat then
                 processed = err
             end
-        end
-
-        if self.sync_chan:get(0.1) ~= nil then
-            log.info("Queue utubettl fiber was stopped")
-            break
+        else
+            -- When switching the master to the replica, the fiber will be stopped.
+            if self.sync_chan:get(0.1) ~= nil then
+                log.info("Queue utubettl fiber was stopped")
+                break
+            end
         end
     end
 end

t/040-utubettl.t

@@ -123,7 +123,7 @@ test:test('ttr put/take', function(test)
     local my_queue = queue.create_tube('trr_test', 'utubettl', { engine = engine })
     test:ok(my_queue:put('ttr1', { ttr = 1 }), 'put ttr task')
     test:ok(my_queue:take(0.1) ~= nil, 'take this task')
-    fiber.sleep(1.5)
+    fiber.sleep(1.1)
     local task = my_queue:peek(0)
     test:is(task[2], state.READY, 'Ready state returned after one second')
 
@@ -132,7 +132,7 @@ test:test('ttr put/take', function(test)
             { engine = engine, storage_mode = queue.driver.utubettl.STORAGE_MODE_READY_BUFFER })
         test:ok(my_queue_ready:put('ttr1', { ttr = 1 }), 'put ttr task')
         test:ok(my_queue_ready:take(0.1) ~= nil, 'take this task')
-        fiber.sleep(1.5)
+        fiber.sleep(1.1)
         local task = my_queue_ready:peek(0)
         test:is(task[2], state.READY, 'Ready state returned after one second')
     end