From d28f22b92f5cdc02431366390105c9b42666c989 Mon Sep 17 00:00:00 2001
From: Oleg Jukovec <oleg.jukovec@tarantool.org>
Date: Thu, 10 Aug 2023 14:19:32 +0300
Subject: [PATCH] drivers: fix concurrency with mvcc

By default idx:max() or idx:min() have read confirmed isolation level.
It could lead to a task duplication or double task take when we
already insert or update a task, commited, but it is not yet
confirmed.

See also:

1. https://github.com/tarantool/queue/issues/207
2. https://github.com/tarantool/queue/pull/211
---
 sharded_queue/drivers/fifo.lua    | 4 ++--
 sharded_queue/drivers/fifottl.lua | 4 ++--
 2 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/sharded_queue/drivers/fifo.lua b/sharded_queue/drivers/fifo.lua
index 1e8f3ef..6a07f13 100644
--- a/sharded_queue/drivers/fifo.lua
+++ b/sharded_queue/drivers/fifo.lua
@@ -74,7 +74,7 @@ end
 
 -- put task in space
 function method.put(args)
-    local task = box.atomic(function()
+    local task = box.atomic({txn_isolation = 'read-committed'}, function()
         local idx = get_index(args)
         local task_id = utils.pack_task_id(
             args.bucket_id,
@@ -96,7 +96,7 @@ end
 
 -- take task
 function method.take(args)
-    local task = box.atomic(function()
+    local task = box.atomic({txn_isolation = 'read-committed'}, function()
         local task = get_space(args).index.status:min { state.READY }
         if task == nil or task[3] ~= state.READY then
             return
diff --git a/sharded_queue/drivers/fifottl.lua b/sharded_queue/drivers/fifottl.lua
index f65097b..ebdb5f4 100644
--- a/sharded_queue/drivers/fifottl.lua
+++ b/sharded_queue/drivers/fifottl.lua
@@ -233,7 +233,7 @@ function method.put(args)
     local ttr = args.ttr or args.options.ttr or ttl
     local priority = args.priority or args.options.priority or 0
 
-    local task = box.atomic(function()
+    local task = box.atomic({txn_isolation = 'read-committed'}, function()
         local idx = get_index(args.tube_name, args.bucket_id)
 
         local next_event
@@ -301,7 +301,7 @@ end
 
 function method.take(args)
 
-    local task = box.atomic(take, args)
+    local task = box.atomic({txn_isolation = 'read-committed'}, take, args)
     if task == nil then return end
 
     if args.extra and args.extra.log_request then