drivers: tolerate a concurrently deleted task on ttl

CHANGELOG.md

@@ -13,6 +13,10 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ### Fixed
 
+- The ttl fiber of the `fifottl` and `utubettl` drivers (and `take()` of
+  `utubettl` in the ready buffer mode) failed with `attempt to index a nil
+  value` when an expired task was deleted concurrently between its
+  selection and `delete()` (gh-264).
 - The `on_task_change` callback got an expired task with a `NULL` status
   instead of `DONE` (`fifottl`, `utubettl`) (#255).
 - `kick()` corrupted the ready buffer of the `utube` driver in the

queue/abstract/driver/fifottl.lua

@@ -113,8 +113,12 @@ local function fifottl_fiber_iteration(self, processed)
         task = self.space.index.watch:min{ task_state }
         if task ~= nil and task[i_status] == task_state then
             if now >= task[i_next_event] then
-                task = self:delete(task[i_id]):transform(2, 1, state.DONE)
-                self:on_task_change(task, 'ttl')
+                -- delete() returns the task with the DONE status or nil if
+                -- the task has been deleted concurrently (gh-264).
+                task = self:delete(task[i_id])
+                if task ~= nil then
+                    self:on_task_change(task, 'ttl')
+                end
                 estimated = 0
                 processed = processed + 1
             else

queue/abstract/driver/utubettl.lua

@@ -240,8 +240,12 @@ local function utubettl_fiber_iteration(self, processed)
         task = self.space.index.watch:min{ task_state }
         if task ~= nil and task[i_status] == task_state then
             if now >= task[i_next_event] then
-                task = self:delete(task[i_id]):transform(2, 1, state.DONE)
-                self:on_task_change(task, 'ttl')
+                -- delete() returns the task with the DONE status or nil if
+                -- the task has been deleted concurrently (gh-264).
+                task = self:delete(task[i_id])
+                if task ~= nil then
+                    self:on_task_change(task, 'ttl')
+                end
                 estimated = 0
                 processed = processed + 1
             else
@@ -518,7 +522,9 @@ local function take_ready(self)
                     take_complete = true
                 end
             else
-                task = self:delete(task[i_id]):transform(2, 1, state.DONE)
+                -- delete() returns the task with the DONE status or nil if
+                -- the task has been deleted concurrently (gh-264).
+                task = self:delete(task[i_id])
                 take_ttl = true
             end
         end
@@ -528,7 +534,7 @@ local function take_ready(self)
         if take_complete then
             self:on_task_change(task, 'take')
             return task
-        elseif take_ttl then
+        elseif take_ttl and task ~= nil then
             self:on_task_change(task, 'ttl')
         end
     end

t/280-ttl-delete-nil.t

@@ -0,0 +1,52 @@
+#!/usr/bin/env tarantool
+local fiber = require('fiber')
+
+local test = require('tap').test('ttl delete nil')
+local queue = require('queue')
+local tnt = require('t.tnt')
+tnt.cfg{}
+
+-- gh-264: the ttl branch of the fiber iteration called :transform() on the
+-- result of delete(), which is nil when the task has been deleted
+-- concurrently between min() and delete(). That killed the ttl fiber.
+local drivers = {'fifottl', 'utubettl'}
+test:plan(#drivers)
+
+for _, driver in ipairs(drivers) do
+    test:test(driver, function(test)
+        test:plan(4)
+        local tube = queue.create_tube(driver .. '_nil', driver)
+        local raw = tube.raw
+        local ttl_fiber = raw.fiber
+
+        -- Emulate a concurrent delete: the first delete() finds no task and
+        -- returns nil, exactly as the driver's delete() does in that case.
+        local calls = 0
+        local delete = raw.delete
+        raw.delete = function(self, id)
+            calls = calls + 1
+            if calls == 1 then
+                self.space:delete(id)
+                return nil
+            end
+            return delete(self, id)
+        end
+
+        tube:put('data', {ttl = 0.1, utube = 'utube'})
+        fiber.sleep(0.3)
+        test:is(calls, 1, 'delete() has returned nil to the ttl fiber')
+        test:isnt(ttl_fiber:status(), 'dead', 'ttl fiber survives a nil from delete()')
+
+        local task = tube:put('data', {ttl = 0.1, utube = 'utube'})
+        fiber.sleep(0.3)
+        test:is(calls, 2, 'delete() is called for the next expired task')
+        test:isnil(raw.space:get(task[1]), 'the next expired task is deleted')
+
+        raw.delete = nil
+        tube:drop()
+    end)
+end
+
+tnt.finish()
+os.exit(test:check() and 0 or 1)
+-- vim: set ft=lua :