From 30b959cb3c6ccffad2c6332233d02fd66ee97ac5 Mon Sep 17 00:00:00 2001 From: Maxim Uymin Date: Tue, 22 Sep 2026 15:24:46 +0300 Subject: [PATCH] drivers: tolerate a concurrently deleted task on ttl The ttl branch of the fiber iteration in fifottl and utubettl (and the expired-task path of utubettl take() in the ready buffer mode) called :transform() on the result of delete(), which is nil when the task has been deleted concurrently between min() and delete(). With vinyl (yielding reads) or MVCC this killed the ttl fiber. The delayed and ttr branches already tolerate a nil from update(). delete() already returns the task with the DONE status, so the extra transform is dropped as well. Closes #264 Co-Authored-By: Claude Fable 5.1 --- CHANGELOG.md | 4 +++ queue/abstract/driver/fifottl.lua | 8 +++-- queue/abstract/driver/utubettl.lua | 14 +++++--- t/280-ttl-delete-nil.t | 52 ++++++++++++++++++++++++++++++ 4 files changed, 72 insertions(+), 6 deletions(-) create mode 100755 t/280-ttl-delete-nil.t diff --git a/CHANGELOG.md b/CHANGELOG.md index 6b82cfb1..75fc7e97 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/queue/abstract/driver/fifottl.lua b/queue/abstract/driver/fifottl.lua index f9ba1db3..cca05b80 100644 --- a/queue/abstract/driver/fifottl.lua +++ b/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 diff --git a/queue/abstract/driver/utubettl.lua b/queue/abstract/driver/utubettl.lua index 3501dbc8..229b9cd9 100644 --- a/queue/abstract/driver/utubettl.lua +++ b/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 diff --git a/t/280-ttl-delete-nil.t b/t/280-ttl-delete-nil.t new file mode 100755 index 00000000..d6fb1de4 --- /dev/null +++ b/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 :