add implementation of the "ttr" queue option

"ttr" in seconds - the time after which, if there is no active
connection in the session, it will be released with all its tasks.

Closes #85

Author: LeonidVas

queue/abstract.lua

@@ -414,15 +414,11 @@ local function make_self(driver, space, tube_name, tube_type, tube_id, opts)
     return self
 end
 
+-- Sessions that don't have any active connections.
+local inactive_sessions = {}
+
 --- Release all session tasks if the session has no actual connections.
 local function cleanup_session_tasks(queue_uuid)
-    local sesion_ids = box.space._queue_session_ids.index.queue_uuid:select(
-        queue_uuid,
-        {limit = 1})[1]
-    if sesion_ids ~= nil then
-        return
-    end
-
     -- Release all session tasks.
     while true do
         local task = box.space._queue_taken.index.pk:min{queue_uuid}
@@ -444,6 +440,26 @@ local function cleanup_session_tasks(queue_uuid)
     end
 end
 
+--- Remove a connection from the list of active connections and
+-- release its tasks if necessary.
+local function remove_connection(box_sid)
+    local queue_session_ids = box.space._queue_session_ids
+    local quuid = get_quuid_by_sid(box_sid)
+
+    queue_session_ids:delete{box_sid}
+    local session_ids = queue_session_ids.index.queue_uuid:select(quuid,
+        {limit = 1})[1]
+
+    if session_ids == nil then
+        local ttr = queue.settings['ttr'] or 0
+        if ttr > 0 then
+            inactive_sessions[quuid] = event_time(ttr)
+        else
+            cleanup_session_tasks(quuid)
+        end
+    end
+end
+
 function method._on_consumer_disconnect()
     local sid = session.id()
 
@@ -465,11 +481,7 @@ function method._on_consumer_disconnect()
         end
     end
 
-    -- Remove connection from the list of active connections and
-    -- release tasks if necessary.
-    local quuid = get_quuid_by_sid(sid)
-    box.space._queue_session_ids:delete{sid}
-    cleanup_session_tasks(quuid)
+    remove_connection(sid)
 end
 
 -- function takes tuples and recreates tube
@@ -574,6 +586,24 @@ local function identification_init()
     end
 end
 
+--- Create an expiration fiber to cleanup expired sessions.
+local function create_expiration_fiber()
+    local exp_fiber = fiber.new(function()
+        while true do
+            local cur_time = time()
+            for quuid, exp_time in pairs(inactive_sessions) do
+                if cur_time >= exp_time then
+                    cleanup_session_tasks(quuid)
+                    inactive_sessions[quuid] = nil
+                end
+            end
+            fiber.sleep(1)
+        end
+    end)
+
+    return exp_fiber
+end
+
 -- create or join infrastructure
 function method.start()
     -- tube_name, tube_id, space_name, tube_type, opts
@@ -667,7 +697,7 @@ function method.start()
     end
 
     identification_init()
-
+    queue.expiration_fiber = create_expiration_fiber()
     session.on_disconnect(queue._on_consumer_disconnect)
     return queue
 end
@@ -746,20 +776,29 @@ function method.identificate(queue_uuid)
         queue_session_ids:insert{sid, quuid}
     elseif queue_uuid ~= nil then
         -- Identificate using a previously created session.
+        -- Check that a session with this uuid exists.
+        local ids_by_uuid = queue_session_ids.index.queue_uuid:select(
+            queue_uuid, {limit = 1})[1]
+        if ids_by_uuid == nil and inactive_sessions[queue_uuid] == nil then
+            error("The UUID " .. uuid.frombin(queue_uuid):str() ..
+                " is unknown.")
+        end
 
         --Validate UUID.
         uuid.frombin(queue_uuid)
 
         if quuid ~= queue_uuid then
-            queue_session_ids:upsert({sid, queue_uuid}, {{'=', 2, queue_uuid}})
             if quuid ~= nil then
-                -- Cleanup old session tasks.
-                cleanup_session_tasks(quuid)
+                remove_connection(sid)
             end
+            queue_session_ids:insert({sid, queue_uuid})
             quuid = queue_uuid
         end
     end
 
+    -- Exclude the session from inactive.
+    inactive_sessions[quuid] = nil
+
     return quuid
 end
 

t/180-work-with-quuid.t

@@ -7,7 +7,7 @@ local os = require('os')
 local tnt  = require('t.tnt')
 
 local test = tap.test('test work with quuid')
-test:plan(1)
+test:plan(3)
 
 local listen = 'localhost:1918'
 tnt.cfg{listen = listen}
@@ -81,5 +81,89 @@ test:test('test work with two consumers with the same quuid', function(test)
     tube:drop()
 end)
 
+test:test('test reconnect and ack the task', function(test)
+    test:plan(2)
+
+    -- Preparing of the tube.
+    queue.set_settings({ttr = 60})
+    local tube = queue.create_tube('test_tube', 'fifo')
+    tube:put('test_data')
+    tube:grant('guest', {call = true})
+
+    local fiber_consumer = fiber.new(function()
+        -- Connect and take a task.
+        local conn = netbox.connect(listen)
+        local task = conn:call('queue.tube.test_tube:take')
+        local task_id = task[1]
+        local session_id = conn:call('queue.identificate')
+        conn:close()
+
+        -- Reconnect and ack the task.
+        conn = netbox.connect(listen)
+        conn:call('queue.identificate', {session_id})
+        task = conn:call('queue.tube.test_tube:ack', {task_id})
+        test:ok(task[1] == task_id and task[2] == '-', 'task has been acked')
+
+        conn:close()
+    end)
+
+    fiber_consumer:set_joinable(true)
+
+    local ok = fiber_consumer:join()
+    test:ok(ok, 'reconnect and ack the task test done')
+
+    tube:drop()
+end)
+
+test:test('test expiration', function(test)
+    test:plan(5)
+
+    -- Preparing of the tube.
+    local tube = queue.create_tube('test_tube', 'fifo')
+    tube:put('test_data')
+    tube:grant('guest', {call = true})
+
+    local fiber_consumer = fiber.new(function()
+        queue.set_settings({ttr = 1})
+        -- Connect and take a task.
+        local conn = netbox.connect(listen)
+        local task_id = conn:call('queue.tube.test_tube:take')[1]
+        local quuid_con = conn:call('queue.identificate')
+        conn:close()
+
+        -- Check that the task is in a "taken" state before ttr expires.
+        fiber.sleep(0.1)
+        test:ok(tube:peek(task_id)[2] == 't', 'task in taken state')
+        fiber.sleep(2)
+
+        -- The task must be released after the ttr expires.
+        test:ok(tube:peek(task_id)[2] == 'r', 'task in ready state after ttr')
+
+        -- The old queue session must expire.
+        conn = netbox.connect(listen)
+        local ok, err = pcall(conn.call, conn, 'queue.identificate',
+            {quuid_con})
+        local check_ident = not ok and err:match('UUID .* is unknown.')
+        test:ok(check_ident, "the old queue session has expired.")
+        conn:close()
+
+        -- If ttr = 0, the task should be released immediately.
+        queue.set_settings({ttr = 0})
+        conn = netbox.connect(listen)
+        task_id = conn:call('queue.tube.test_tube:take')[1]
+        conn:close()
+        fiber.sleep(0.1)
+        test:ok(tube:peek(task_id)[2] == 'r',
+            'task has been immediately released')
+    end)
+
+    fiber_consumer:set_joinable(true)
+
+    local ok = fiber_consumer:join()
+    test:ok(ok, 'expiration test done')
+
+    tube:drop()
+end)
+
 tnt.finish()
 os.exit(test:check() and 0 or 1)